From 69ac454f29058be9f9175d57ec8cad2eff5022eb Mon Sep 17 00:00:00 2001 From: David Kaya Date: Sun, 5 Apr 2026 20:22:20 +0200 Subject: [PATCH] feat(workflows): add phase 4 backend executor support Add MVP code-executor and function-executor runtime support, request-port bridging through Aryx user input, state-scope helpers, lockstep execution mode handling, and validation/tests for new workflow node kinds. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Services/CopilotUserInputCoordinator.cs | 106 ++- .../Services/CopilotWorkflowRunner.cs | 215 +++++- .../Services/WorkflowHostExecutors.cs | 721 +++++++++++++++++- .../Aryx.AgentHost/Services/WorkflowRunner.cs | 115 ++- .../Services/WorkflowValidator.cs | 79 +- .../CopilotWorkflowRunnerTests.cs | 159 ++++ .../SidecarProtocolHostTests.cs | 2 +- .../WorkflowRunnerTests.cs | 316 ++++++++ .../WorkflowValidatorTests.cs | 125 +++ src/shared/domain/workflow.ts | 79 +- tests/shared/workflow.test.ts | 102 ++- 11 files changed, 1943 insertions(+), 76 deletions(-) diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotUserInputCoordinator.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotUserInputCoordinator.cs index c7df86e..089c98f 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotUserInputCoordinator.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotUserInputCoordinator.cs @@ -44,31 +44,32 @@ internal sealed class CopilotUserInputCoordinator ArgumentNullException.ThrowIfNull(invocation); ArgumentNullException.ThrowIfNull(onUserInput); - PendingUserInputRequest pending = CreatePendingUserInput(command); - if (!_pendingUserInputs.TryAdd(pending.UserInputId, pending)) - { - throw new InvalidOperationException($"User input request \"{pending.UserInputId}\" is already pending."); - } + return await RequestUserInputCoreAsync( + command, + agent.Id, + agent.Name, + request, + onUserInput, + cancellationToken).ConfigureAwait(false); + } - try - { - await onUserInput(BuildUserInputRequestedEvent(command, agent, request, pending.UserInputId)) - .ConfigureAwait(false); + public Task RequestUserInputAsync( + RunTurnCommandDto command, + UserInputRequest request, + Func onUserInput, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(command); + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(onUserInput); - using CancellationTokenRegistration registration = cancellationToken.Register( - static state => - { - ((TaskCompletionSource)state!) - .TrySetCanceled(); - }, - pending.Response); - - return await pending.Response.Task.ConfigureAwait(false); - } - finally - { - _pendingUserInputs.TryRemove(pending.UserInputId, out _); - } + return RequestUserInputCoreAsync( + command, + agentId: null, + agentName: null, + request, + onUserInput, + cancellationToken); } internal static UserInputRequestedEventDto BuildUserInputRequestedEvent( @@ -98,6 +99,65 @@ internal sealed class CopilotUserInputCoordinator }; } + internal static UserInputRequestedEventDto BuildUserInputRequestedEvent( + RunTurnCommandDto command, + string? agentId, + string? agentName, + UserInputRequest request, + string userInputId) + { + ArgumentNullException.ThrowIfNull(command); + ArgumentNullException.ThrowIfNull(request); + + return new UserInputRequestedEventDto + { + Type = "user-input-requested", + RequestId = command.RequestId, + SessionId = command.SessionId, + UserInputId = userInputId, + AgentId = NormalizeOptionalString(agentId), + AgentName = NormalizeOptionalString(agentName), + Question = NormalizeOptionalString(request.Question) ?? string.Empty, + Choices = NormalizeOptionalStringList(request.Choices ?? []), + AllowFreeform = request.AllowFreeform, + }; + } + + private async Task RequestUserInputCoreAsync( + RunTurnCommandDto command, + string? agentId, + string? agentName, + UserInputRequest request, + Func onUserInput, + CancellationToken cancellationToken) + { + PendingUserInputRequest pending = CreatePendingUserInput(command); + if (!_pendingUserInputs.TryAdd(pending.UserInputId, pending)) + { + throw new InvalidOperationException($"User input request \"{pending.UserInputId}\" is already pending."); + } + + try + { + await onUserInput(BuildUserInputRequestedEvent(command, agentId, agentName, request, pending.UserInputId)) + .ConfigureAwait(false); + + using CancellationTokenRegistration registration = cancellationToken.Register( + static state => + { + ((TaskCompletionSource)state!) + .TrySetCanceled(); + }, + pending.Response); + + return await pending.Response.Task.ConfigureAwait(false); + } + finally + { + _pendingUserInputs.TryRemove(pending.UserInputId, out _); + } + } + private static PendingUserInputRequest CreatePendingUserInput(RunTurnCommandDto command) { return new PendingUserInputRequest( diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs index e01ab32..6a5ebdf 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs @@ -1,9 +1,12 @@ 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.Workflows; using Microsoft.Agents.AI.Workflows.Checkpointing; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Extensions.AI; namespace Aryx.AgentHost.Services; @@ -104,6 +107,17 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner 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); @@ -179,24 +193,32 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner IReadOnlyList inputMessages, CheckpointManager? checkpointManager) { + InProcessExecutionEnvironment environment = CreateExecutionEnvironment(command, checkpointManager); if (checkpointManager is not null && command.ResumeFromCheckpoint is { } resumeFromCheckpoint) { - return InProcessExecution.ResumeStreamingAsync( + return environment.ResumeStreamingAsync( workflow, - new CheckpointInfo(resumeFromCheckpoint.WorkflowSessionId, resumeFromCheckpoint.CheckpointId), - checkpointManager); + new CheckpointInfo(resumeFromCheckpoint.WorkflowSessionId, resumeFromCheckpoint.CheckpointId)); } - if (checkpointManager is not null) - { - return InProcessExecution.RunStreamingAsync( - workflow, - inputMessages.ToList(), - checkpointManager, - sessionId: command.RequestId); - } + return environment.RunStreamingAsync(workflow, inputMessages.ToList(), command.RequestId); + } - return InProcessExecution.RunStreamingAsync(workflow, inputMessages.ToList()); + 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( @@ -243,6 +265,30 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner 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( RunTurnCommandDto command, WorkflowEvent evt, @@ -336,6 +382,74 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner 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, @@ -543,6 +657,83 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner } } + 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( diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs index 6220aa6..2e54d4b 100644 --- a/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs @@ -1,25 +1,69 @@ +using System.Globalization; +using System.Text.Json; +using Aryx.AgentHost.Contracts; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Workflows; using Microsoft.Extensions.AI; namespace Aryx.AgentHost.Services; -internal sealed class WorkflowOutputMessagesExecutor(ChatProtocolExecutorOptions? options = null) - : ChatProtocolExecutor(ExecutorId, options, declareCrossRunShareable: true), IResettableExecutor +internal sealed class WorkflowOutputMessagesExecutor(string id = "OutputMessages") + : Executor(id, declareCrossRunShareable: true), IResettableExecutor { public const string ExecutorId = "OutputMessages"; protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) - => base.ConfigureProtocol(protocolBuilder) + { + return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder + .AddHandler(static (_, _, _) => default) + .AddHandler(YieldMessageAsync) + .AddHandler>(YieldMessagesAsync) + .AddHandler(YieldMessageArrayAsync) + .AddHandler>(YieldEnumerableMessagesAsync) + .AddCatchAll(YieldCatchAllAsync)) .YieldsOutput>(); + } - protected override ValueTask TakeTurnAsync( + private static ValueTask YieldMessageAsync( + ChatMessage message, + IWorkflowContext context, + CancellationToken cancellationToken) + => context.YieldOutputAsync(new List { message }, cancellationToken); + + private static ValueTask YieldMessagesAsync( List messages, IWorkflowContext context, - bool? emitEvents, - CancellationToken cancellationToken = default) + CancellationToken cancellationToken) => context.YieldOutputAsync(messages, cancellationToken); + private static ValueTask YieldMessageArrayAsync( + ChatMessage[] messages, + IWorkflowContext context, + CancellationToken cancellationToken) + => context.YieldOutputAsync(messages.ToList(), cancellationToken); + + private static ValueTask YieldEnumerableMessagesAsync( + IEnumerable messages, + IWorkflowContext context, + CancellationToken cancellationToken) + => context.YieldOutputAsync(messages.ToList(), cancellationToken); + + private static ValueTask YieldCatchAllAsync( + PortableValue message, + IWorkflowContext context, + CancellationToken cancellationToken) + { + if (message.Is()) + { + return default; + } + + object payload = message.As() ?? WorkflowValueSerializer.CreateNullElement(); + return context.YieldOutputAsync( + WorkflowValueSerializer.ToOutputMessages(payload), + cancellationToken); + } + ValueTask IResettableExecutor.ResetAsync() => default; } @@ -147,3 +191,668 @@ internal sealed class WorkflowRoundRobinGroupChatHost( ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync(); } + +internal sealed class WorkflowStateScopeCatalog +{ + public static WorkflowStateScopeCatalog Empty { get; } = new([]); + + private readonly IReadOnlyDictionary> _scopes; + + public WorkflowStateScopeCatalog(IReadOnlyList? stateScopes) + { + Dictionary> scopes = new(StringComparer.OrdinalIgnoreCase); + foreach (WorkflowStateScopeDto scope in stateScopes ?? []) + { + string? scopeName = NormalizeOptionalString(scope.Name); + if (scopeName is null) + { + continue; + } + + Dictionary initialValues = new(StringComparer.OrdinalIgnoreCase); + foreach ((string key, JsonElement value) in scope.InitialValues + ?? new Dictionary(StringComparer.OrdinalIgnoreCase)) + { + string? normalizedKey = NormalizeOptionalString(key); + if (normalizedKey is null) + { + continue; + } + + initialValues[normalizedKey] = WorkflowValueSerializer.CloneElement(value); + } + + scopes[scopeName] = initialValues; + } + + _scopes = scopes; + } + + public async ValueTask ReadJsonStateAsync( + IWorkflowContext context, + string scopeName, + string key, + CancellationToken cancellationToken) + { + string normalizedScope = NormalizeRequired(scopeName, nameof(scopeName)); + string normalizedKey = NormalizeRequired(key, nameof(key)); + + if (TryGetInitialValue(normalizedScope, normalizedKey, out JsonElement initialValue)) + { + JsonElement value = await context.ReadOrInitStateAsync( + normalizedKey, + () => WorkflowValueSerializer.CloneElement(initialValue), + normalizedScope, + cancellationToken).ConfigureAwait(false); + return WorkflowValueSerializer.CloneElement(value); + } + + JsonElement? existing = await context.ReadStateAsync( + normalizedKey, + normalizedScope, + cancellationToken).ConfigureAwait(false); + return existing.HasValue ? WorkflowValueSerializer.CloneElement(existing.Value) : null; + } + + public ValueTask QueueJsonStateUpdateAsync( + IWorkflowContext context, + string scopeName, + string key, + JsonElement value, + CancellationToken cancellationToken) + { + string normalizedScope = NormalizeRequired(scopeName, nameof(scopeName)); + string normalizedKey = NormalizeRequired(key, nameof(key)); + return context.QueueStateUpdateAsync( + normalizedKey, + WorkflowValueSerializer.CloneElement(value), + normalizedScope, + cancellationToken); + } + + private bool TryGetInitialValue(string scopeName, string key, out JsonElement value) + { + value = default; + return _scopes.TryGetValue(scopeName, out IReadOnlyDictionary? scope) + && scope.TryGetValue(key, out value); + } + + private static string NormalizeRequired(string value, string paramName) + { + return NormalizeOptionalString(value) + ?? throw new InvalidOperationException($"{paramName} is required."); + } + + private static string? NormalizeOptionalString(string? value) + => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); +} + +internal sealed record WorkflowRequestPortNodeDefinition( + string NodeId, + string NodeLabel, + string PortId, + string RequestType, + string ResponseType, + string? Prompt); + +internal sealed class WorkflowRequestPortPromptRequest +{ + public string NodeId { get; init; } = string.Empty; + + public string NodeLabel { get; init; } = string.Empty; + + public string PortId { get; init; } = string.Empty; + + public string RequestType { get; init; } = string.Empty; + + public string ResponseType { get; init; } = string.Empty; + + public string? Prompt { get; init; } + + public string? InputSummary { get; init; } +} + +internal sealed class WorkflowCodeExecutor( + string id, + string implementation, + WorkflowStateScopeCatalog stateCatalog) + : Executor(id, declareCrossRunShareable: true), IResettableExecutor +{ + private readonly string _implementation = implementation; + private readonly WorkflowStateScopeCatalog _stateCatalog = stateCatalog; + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + { + return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder + .AddHandler(static (_, _, _) => default) + .AddCatchAll(HandleAsync)) + .SendsMessage(); + } + + private async ValueTask HandleAsync( + PortableValue message, + IWorkflowContext context, + CancellationToken cancellationToken) + { + object input = message.As() ?? WorkflowValueSerializer.CreateNullElement(); + object result = await ExecuteAsync(input, context, cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(result, cancellationToken: cancellationToken).ConfigureAwait(false); + } + + private ValueTask ExecuteAsync( + object input, + IWorkflowContext context, + CancellationToken cancellationToken) + { + if (string.Equals(_implementation, "return-input", StringComparison.OrdinalIgnoreCase)) + { + return ValueTask.FromResult(input); + } + + if (_implementation.StartsWith("return-text:", StringComparison.OrdinalIgnoreCase)) + { + return ValueTask.FromResult(_implementation["return-text:".Length..]); + } + + if (_implementation.StartsWith("return-json:", StringComparison.OrdinalIgnoreCase)) + { + string rawJson = _implementation["return-json:".Length..]; + return ValueTask.FromResult(WorkflowValueSerializer.ParseJsonElement(rawJson)); + } + + if (_implementation.StartsWith("state:set:", StringComparison.OrdinalIgnoreCase)) + { + return ExecuteStateSetAsync( + _implementation.Split(':', 5), + context, + cancellationToken); + } + + if (_implementation.StartsWith("state:get:", StringComparison.OrdinalIgnoreCase)) + { + return ExecuteStateGetAsync( + _implementation.Split(':', 4), + context, + cancellationToken); + } + + throw new InvalidOperationException( + $"Code executor \"{Id}\" does not support implementation \"{_implementation}\". " + + "Supported implementations are return-input, return-text:, return-json:, state:set:::, and state:get::."); + } + + private async ValueTask ExecuteStateSetAsync( + string[] segments, + IWorkflowContext context, + CancellationToken cancellationToken) + { + if (segments.Length != 5) + { + throw new InvalidOperationException( + $"Code executor \"{Id}\" requires the format state:set:::. Received \"{_implementation}\"."); + } + + JsonElement value = WorkflowValueSerializer.ParseJsonElement(segments[4]); + await _stateCatalog.QueueJsonStateUpdateAsync( + context, + segments[2], + segments[3], + value, + cancellationToken).ConfigureAwait(false); + return value; + } + + private async ValueTask ExecuteStateGetAsync( + string[] segments, + IWorkflowContext context, + CancellationToken cancellationToken) + { + if (segments.Length != 4) + { + throw new InvalidOperationException( + $"Code executor \"{Id}\" requires the format state:get::. Received \"{_implementation}\"."); + } + + JsonElement? value = await _stateCatalog.ReadJsonStateAsync( + context, + segments[2], + segments[3], + cancellationToken).ConfigureAwait(false); + return value ?? WorkflowValueSerializer.CreateNullElement(); + } + + public ValueTask ResetAsync() => default; +} + +internal sealed class WorkflowFunctionExecutor( + string id, + string functionRef, + IReadOnlyDictionary? parameters, + WorkflowStateScopeCatalog stateCatalog) + : Executor(id, declareCrossRunShareable: true), IResettableExecutor +{ + private readonly string _functionRef = functionRef; + private readonly IReadOnlyDictionary _parameters = parameters ?? new Dictionary(StringComparer.OrdinalIgnoreCase); + private readonly WorkflowStateScopeCatalog _stateCatalog = stateCatalog; + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + { + return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder + .AddHandler(static (_, _, _) => default) + .AddCatchAll(HandleAsync)) + .SendsMessage(); + } + + private async ValueTask HandleAsync( + PortableValue message, + IWorkflowContext context, + CancellationToken cancellationToken) + { + object input = message.As() ?? WorkflowValueSerializer.CreateNullElement(); + object result = await WorkflowFunctionRegistry.InvokeAsync( + _functionRef, + input, + _parameters, + context, + _stateCatalog, + cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(result, cancellationToken: cancellationToken).ConfigureAwait(false); + } + + public ValueTask ResetAsync() => default; +} + +internal static class WorkflowFunctionRegistry +{ + private static readonly HashSet SupportedFunctionRefs = new(StringComparer.OrdinalIgnoreCase) + { + "identity", + "return-parameter", + "concat-text", + "state:get", + "state:set", + }; + + public static bool IsSupported(string? functionRef) + => !string.IsNullOrWhiteSpace(functionRef) && SupportedFunctionRefs.Contains(functionRef.Trim()); + + public static async ValueTask InvokeAsync( + string functionRef, + object input, + IReadOnlyDictionary parameters, + IWorkflowContext context, + WorkflowStateScopeCatalog stateCatalog, + CancellationToken cancellationToken) + { + string normalizedFunctionRef = functionRef.Trim(); + return normalizedFunctionRef switch + { + var value when string.Equals(value, "identity", StringComparison.OrdinalIgnoreCase) + => input, + var value when string.Equals(value, "return-parameter", StringComparison.OrdinalIgnoreCase) + => ReturnParameter(parameters), + var value when string.Equals(value, "concat-text", StringComparison.OrdinalIgnoreCase) + => ConcatText(input, parameters), + var value when string.Equals(value, "state:get", StringComparison.OrdinalIgnoreCase) + => await GetStateAsync(parameters, context, stateCatalog, cancellationToken).ConfigureAwait(false), + var value when string.Equals(value, "state:set", StringComparison.OrdinalIgnoreCase) + => await SetStateAsync(parameters, context, stateCatalog, cancellationToken).ConfigureAwait(false), + _ => throw new InvalidOperationException( + $"Function executor references unsupported functionRef \"{functionRef}\". Supported refs are: {string.Join(", ", SupportedFunctionRefs.OrderBy(static value => value, StringComparer.OrdinalIgnoreCase))}.") + }; + } + + private static object ReturnParameter(IReadOnlyDictionary parameters) + { + if (TryGetParameter(parameters, "name", out JsonElement namedParameterSelector) + && namedParameterSelector.ValueKind == JsonValueKind.String) + { + string parameterName = namedParameterSelector.GetString() ?? string.Empty; + if (TryGetParameter(parameters, parameterName, out JsonElement namedValue)) + { + return WorkflowValueSerializer.CloneElement(namedValue); + } + + throw new InvalidOperationException( + $"Function executor return-parameter could not find parameter \"{parameterName}\"."); + } + + if (TryGetParameter(parameters, "value", out JsonElement value)) + { + return WorkflowValueSerializer.CloneElement(value); + } + + KeyValuePair[] remaining = parameters + .Where(static pair => !string.Equals(pair.Key, "name", StringComparison.OrdinalIgnoreCase)) + .ToArray(); + if (remaining.Length == 1) + { + return WorkflowValueSerializer.CloneElement(remaining[0].Value); + } + + throw new InvalidOperationException( + "Function executor return-parameter requires either a value parameter, a name selector, or exactly one parameter value."); + } + + private static object ConcatText(object input, IReadOnlyDictionary parameters) + { + List parts = []; + if (TryGetString(parameters, "prefix", out string? prefix)) + { + parts.Add(prefix!); + } + + bool includeInput = !TryGetBoolean(parameters, "includeInput", out bool parsedIncludeInput) || parsedIncludeInput; + if (includeInput) + { + parts.Add(WorkflowValueSerializer.ToDisplayText(input)); + } + + if (TryGetParameter(parameters, "values", out JsonElement values)) + { + if (values.ValueKind != JsonValueKind.Array) + { + throw new InvalidOperationException("Function executor concat-text requires values to be a JSON array when provided."); + } + + foreach (JsonElement element in values.EnumerateArray()) + { + parts.Add(WorkflowValueSerializer.ToDisplayText(element)); + } + } + + if (TryGetString(parameters, "suffix", out string? suffix)) + { + parts.Add(suffix!); + } + + string separator = TryGetString(parameters, "separator", out string? parsedSeparator) + ? parsedSeparator! + : string.Empty; + return string.Join(separator, parts); + } + + private static async ValueTask GetStateAsync( + IReadOnlyDictionary parameters, + IWorkflowContext context, + WorkflowStateScopeCatalog stateCatalog, + CancellationToken cancellationToken) + { + string scope = GetRequiredString(parameters, "scope"); + string key = GetRequiredString(parameters, "key"); + JsonElement? value = await stateCatalog.ReadJsonStateAsync( + context, + scope, + key, + cancellationToken).ConfigureAwait(false); + return value ?? WorkflowValueSerializer.CreateNullElement(); + } + + private static async ValueTask SetStateAsync( + IReadOnlyDictionary parameters, + IWorkflowContext context, + WorkflowStateScopeCatalog stateCatalog, + CancellationToken cancellationToken) + { + string scope = GetRequiredString(parameters, "scope"); + string key = GetRequiredString(parameters, "key"); + JsonElement value = GetRequiredJson(parameters, "value"); + await stateCatalog.QueueJsonStateUpdateAsync( + context, + scope, + key, + value, + cancellationToken).ConfigureAwait(false); + return value; + } + + private static string GetRequiredString(IReadOnlyDictionary parameters, string name) + { + if (TryGetString(parameters, name, out string? value) && !string.IsNullOrWhiteSpace(value)) + { + return value; + } + + throw new InvalidOperationException($"Function executor requires a non-empty string parameter \"{name}\"."); + } + + private static JsonElement GetRequiredJson(IReadOnlyDictionary parameters, string name) + { + if (TryGetParameter(parameters, name, out JsonElement value)) + { + return WorkflowValueSerializer.CloneElement(value); + } + + throw new InvalidOperationException($"Function executor requires parameter \"{name}\"."); + } + + private static bool TryGetString(IReadOnlyDictionary parameters, string name, out string? value) + { + value = null; + if (!TryGetParameter(parameters, name, out JsonElement element)) + { + return false; + } + + value = element.ValueKind switch + { + JsonValueKind.String => element.GetString(), + JsonValueKind.True => bool.TrueString, + JsonValueKind.False => bool.FalseString, + JsonValueKind.Number => element.ToString(), + _ => null, + }; + return value is not null; + } + + private static bool TryGetBoolean(IReadOnlyDictionary parameters, string name, out bool value) + { + value = false; + if (!TryGetParameter(parameters, name, out JsonElement element)) + { + return false; + } + + if (element.ValueKind == JsonValueKind.True || element.ValueKind == JsonValueKind.False) + { + value = element.GetBoolean(); + return true; + } + + if (element.ValueKind == JsonValueKind.String && bool.TryParse(element.GetString(), out bool parsed)) + { + value = parsed; + return true; + } + + return false; + } + + private static bool TryGetParameter(IReadOnlyDictionary parameters, string name, out JsonElement value) + { + foreach ((string key, JsonElement parameterValue) in parameters) + { + if (string.Equals(key, name, StringComparison.OrdinalIgnoreCase)) + { + value = parameterValue; + return true; + } + } + + value = default; + return false; + } +} + +internal sealed class WorkflowRequestPortIngressExecutor( + WorkflowRequestPortNodeDefinition definition, + RequestPort port) + : Executor($"{definition.NodeId}::request-entry", declareCrossRunShareable: true), IResettableExecutor +{ + private readonly WorkflowRequestPortNodeDefinition _definition = definition; + private readonly RequestPort _port = port; + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + { + return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder + .AddHandler(static (_, _, _) => default) + .AddCatchAll(HandleAsync)) + .SendsMessage(); + } + + private async ValueTask HandleAsync( + PortableValue message, + IWorkflowContext context, + CancellationToken cancellationToken) + { + object input = message.As() ?? WorkflowValueSerializer.CreateNullElement(); + WorkflowRequestPortPromptRequest request = new() + { + NodeId = _definition.NodeId, + NodeLabel = _definition.NodeLabel, + PortId = _definition.PortId, + RequestType = _definition.RequestType, + ResponseType = _definition.ResponseType, + Prompt = _definition.Prompt, + InputSummary = WorkflowValueSerializer.ToPromptSummary(input), + }; + + await context.SendMessageAsync(request, _port.Id, cancellationToken).ConfigureAwait(false); + } + + public ValueTask ResetAsync() => default; +} + +internal sealed class WorkflowRequestPortResponseExecutor(string nodeId) + : Executor($"{nodeId}::request-exit", declareCrossRunShareable: true), IResettableExecutor +{ + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + { + return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder + .AddHandler(static (_, _, _) => default) + .AddHandler(static (_, _, _) => default) + .AddCatchAll(ForwardAsync)) + .SendsMessage(); + } + + private static ValueTask ForwardAsync( + PortableValue message, + IWorkflowContext context, + CancellationToken cancellationToken) + { + object payload = message.As() ?? WorkflowValueSerializer.CreateNullElement(); + return context.SendMessageAsync(payload, cancellationToken: cancellationToken); + } + + public ValueTask ResetAsync() => default; +} + +internal static class WorkflowValueSerializer +{ + private static readonly JsonSerializerOptions JsonOptions = JsonSerialization.CreateWebOptions(); + + public static JsonElement CloneElement(JsonElement value) => value.Clone(); + + public static JsonElement ParseJsonElement(string json) + { + try + { + return JsonDocument.Parse(json).RootElement.Clone(); + } + catch (JsonException ex) + { + throw new InvalidOperationException($"Invalid JSON payload: {ex.Message}", ex); + } + } + + public static JsonElement CreateNullElement() => JsonDocument.Parse("null").RootElement.Clone(); + + public static List ToOutputMessages(object value) + { + if (value is List chatMessages) + { + return chatMessages; + } + + if (value is ChatMessage chatMessage) + { + return [chatMessage]; + } + + if (value is ChatMessage[] chatMessageArray) + { + return [.. chatMessageArray]; + } + + if (value is IEnumerable enumerable) + { + return enumerable.ToList(); + } + + return [ + new ChatMessage(ChatRole.Assistant, ToDisplayText(value)) + { + AuthorName = "Workflow", + }, + ]; + } + + public static string ToDisplayText(object? value) + { + if (value is null) + { + return "null"; + } + + if (value is string text) + { + return text; + } + + if (value is JsonElement jsonElement) + { + return jsonElement.ValueKind switch + { + JsonValueKind.String => jsonElement.GetString() ?? string.Empty, + JsonValueKind.True => bool.TrueString, + JsonValueKind.False => bool.FalseString, + JsonValueKind.Number => jsonElement.ToString(), + JsonValueKind.Null => "null", + _ => jsonElement.GetRawText(), + }; + } + + if (value is ChatMessage chatMessage) + { + return chatMessage.Text ?? string.Empty; + } + + if (value is IEnumerable messages) + { + return string.Join(Environment.NewLine, messages.Select(static message => message.Text ?? string.Empty)); + } + + if (value is bool boolean) + { + return boolean ? bool.TrueString : bool.FalseString; + } + + if (value is IFormattable formattable) + { + return formattable.ToString(null, CultureInfo.InvariantCulture); + } + + return JsonSerializer.Serialize(value, JsonOptions); + } + + public static string? ToPromptSummary(object? value) + { + if (value is null || value is JsonElement jsonElement && jsonElement.ValueKind == JsonValueKind.Null) + { + return null; + } + + string summary = ToDisplayText(value); + return string.IsNullOrWhiteSpace(summary) ? null : summary; + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs index 80dafa4..8f98c76 100644 --- a/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs @@ -22,11 +22,6 @@ internal sealed class WorkflowRunner .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal) ?? new Dictionary(StringComparer.Ordinal); - WorkflowNodeDto startNode = workflowDefinition.Graph.Nodes.Single(node => - string.Equals(node.Kind, "start", StringComparison.OrdinalIgnoreCase)); - WorkflowNodeDto endNode = workflowDefinition.Graph.Nodes.Single(node => - string.Equals(node.Kind, "end", StringComparison.OrdinalIgnoreCase)); - Dictionary agentMap = patternDefinition.Agents .Zip(agents, (definition, agent) => (definition.Id, agent)) .ToDictionary(pair => pair.Id, pair => pair.agent, StringComparer.Ordinal); @@ -43,26 +38,37 @@ internal sealed class WorkflowRunner string.Equals(node.Kind, "start", StringComparison.OrdinalIgnoreCase)); WorkflowNodeDto endNode = workflowDefinition.Graph.Nodes.Single(node => string.Equals(node.Kind, "end", StringComparison.OrdinalIgnoreCase)); + WorkflowStateScopeCatalog stateCatalog = new(workflowDefinition.Settings.StateScopes); - Dictionary bindings = new(StringComparer.Ordinal); + Dictionary routes = new(StringComparer.Ordinal); foreach (WorkflowNodeDto node in workflowDefinition.Graph.Nodes) { - bindings[node.Id] = CreateExecutorBinding(node, agentMap, workflowLibrary); + routes[node.Id] = CreateNodeRoute(node, agentMap, workflowLibrary, stateCatalog); } - WorkflowBuilder builder = new(bindings[startNode.Id]); + WorkflowBuilder builder = new(routes[startNode.Id].Entry); + + foreach (WorkflowNodeRoute route in routes.Values) + { + foreach ((ExecutorBinding source, ExecutorBinding target) in route.InternalEdges) + { + builder.AddEdge(source, target); + } + } foreach (WorkflowEdgeDto edge in workflowDefinition.Graph.Edges.Where(edge => string.Equals(edge.Kind, "direct", StringComparison.OrdinalIgnoreCase))) { Func? condition = WorkflowConditionEvaluator.Compile(edge); + ExecutorBinding source = routes[edge.Source].Exit; + ExecutorBinding target = routes[edge.Target].Entry; if (condition is null) { - builder.AddEdge(bindings[edge.Source], bindings[edge.Target]); + builder.AddEdge(source, target); } else { - builder.AddEdge(bindings[edge.Source], bindings[edge.Target], condition); + builder.AddEdge(source, target, condition); } } @@ -71,19 +77,20 @@ internal sealed class WorkflowRunner .GroupBy(edge => edge.Source, StringComparer.Ordinal)) { WorkflowEdgeDto[] fanOutEdges = fanOutGroup.ToArray(); - ExecutorBinding[] targets = fanOutEdges.Select(edge => bindings[edge.Target]).ToArray(); + ExecutorBinding source = routes[fanOutGroup.Key].Exit; + ExecutorBinding[] targets = fanOutEdges.Select(edge => routes[edge.Target].Entry).ToArray(); Func?[] compiledConditions = fanOutEdges .Select(WorkflowConditionEvaluator.Compile) .ToArray(); bool hasConditionalRouting = fanOutEdges.Any(edge => edge.Condition is not null); if (!hasConditionalRouting) { - builder.AddFanOutEdge(bindings[fanOutGroup.Key], targets); + builder.AddFanOutEdge(source, targets); continue; } builder.AddFanOutEdge( - bindings[fanOutGroup.Key], + source, targets, (payload, _) => fanOutEdges .Select((edge, index) => (edge, index)) @@ -97,8 +104,8 @@ internal sealed class WorkflowRunner .GroupBy(edge => edge.Target, StringComparer.Ordinal)) { builder.AddFanInBarrierEdge( - fanInGroup.Select(edge => bindings[edge.Source]).ToArray(), - bindings[fanInGroup.Key]); + fanInGroup.Select(edge => routes[edge.Source].Exit).ToArray(), + routes[fanInGroup.Key].Entry); } if (!string.IsNullOrWhiteSpace(workflowDefinition.Name)) @@ -106,22 +113,25 @@ internal sealed class WorkflowRunner builder = builder.WithName(workflowDefinition.Name); } - return builder.WithOutputFrom(bindings[endNode.Id]).Build(); + return builder.WithOutputFrom(routes[endNode.Id].Exit).Build(); } - private static ExecutorBinding CreateExecutorBinding( + private WorkflowNodeRoute CreateNodeRoute( WorkflowNodeDto node, IReadOnlyDictionary agentMap, - IReadOnlyDictionary workflowLibrary) + IReadOnlyDictionary workflowLibrary, + WorkflowStateScopeCatalog stateCatalog) { if (string.Equals(node.Kind, "start", StringComparison.OrdinalIgnoreCase)) { - return new ChatForwardingExecutor(node.Id); + ExecutorBinding binding = new ChatForwardingExecutor(node.Id).BindExecutor(); + return new WorkflowNodeRoute(binding); } if (string.Equals(node.Kind, "end", StringComparison.OrdinalIgnoreCase)) { - return new WorkflowOutputMessagesExecutor(); + ExecutorBinding binding = new WorkflowOutputMessagesExecutor(node.Id).BindExecutor(); + return new WorkflowNodeRoute(binding); } if (string.Equals(node.Kind, "agent", StringComparison.OrdinalIgnoreCase)) @@ -132,19 +142,61 @@ internal sealed class WorkflowRunner throw new InvalidOperationException($"Workflow node \"{node.Id}\" references unknown agent \"{agentId}\"."); } - return agent.BindAsExecutor(CopilotAgentBundle.CreateAgentHostOptions()); + return new WorkflowNodeRoute(agent.BindAsExecutor(CopilotAgentBundle.CreateAgentHostOptions())); + } + + if (string.Equals(node.Kind, "code-executor", StringComparison.OrdinalIgnoreCase)) + { + string implementation = NormalizeRequired(node.Config.Implementation, $"Workflow code executor \"{node.Id}\" requires an implementation."); + ExecutorBinding binding = new WorkflowCodeExecutor(node.Id, implementation, stateCatalog).BindExecutor(); + return new WorkflowNodeRoute(binding); + } + + if (string.Equals(node.Kind, "function-executor", StringComparison.OrdinalIgnoreCase)) + { + string functionRef = NormalizeRequired(node.Config.FunctionRef, $"Workflow function executor \"{node.Id}\" requires a functionRef."); + if (!WorkflowFunctionRegistry.IsSupported(functionRef)) + { + throw new InvalidOperationException( + $"Workflow function executor \"{node.Id}\" references unsupported functionRef \"{functionRef}\"."); + } + + ExecutorBinding binding = new WorkflowFunctionExecutor(node.Id, functionRef, node.Config.Parameters, stateCatalog).BindExecutor(); + return new WorkflowNodeRoute(binding); + } + + if (string.Equals(node.Kind, "request-port", StringComparison.OrdinalIgnoreCase)) + { + return CreateRequestPortRoute(node); } if (string.Equals(node.Kind, "sub-workflow", StringComparison.OrdinalIgnoreCase)) { WorkflowDefinitionDto subWorkflowDefinition = ResolveSubWorkflowDefinition(node, workflowLibrary); - Workflow subWorkflow = new WorkflowRunner().BuildWorkflow(subWorkflowDefinition, agentMap, workflowLibrary); - return subWorkflow.BindAsExecutor(node.Id); + Workflow subWorkflow = BuildWorkflow(subWorkflowDefinition, agentMap, workflowLibrary); + return new WorkflowNodeRoute(subWorkflow.BindAsExecutor(node.Id)); } throw new NotSupportedException($"Workflow node kind \"{node.Kind}\" is not executable yet."); } + private static WorkflowNodeRoute CreateRequestPortRoute(WorkflowNodeDto node) + { + WorkflowRequestPortNodeDefinition definition = new( + node.Id, + NormalizeOptionalString(node.Label) ?? node.Id, + NormalizeRequired(node.Config.PortId, $"Workflow request port \"{node.Id}\" requires a portId."), + NormalizeRequired(node.Config.RequestType, $"Workflow request port \"{node.Id}\" requires a requestType."), + NormalizeRequired(node.Config.ResponseType, $"Workflow request port \"{node.Id}\" requires a responseType."), + NormalizeOptionalString(node.Config.Prompt)); + + RequestPort port = new(definition.PortId, typeof(WorkflowRequestPortPromptRequest), typeof(object)); + ExecutorBinding entry = new WorkflowRequestPortIngressExecutor(definition, port).BindExecutor(); + ExecutorBinding portBinding = new RequestPortBinding(port, false); + ExecutorBinding exit = new WorkflowRequestPortResponseExecutor(node.Id).BindExecutor(); + return new WorkflowNodeRoute(entry, exit, [(entry, portBinding), (portBinding, exit)]); + } + private static WorkflowDefinitionDto ResolveSubWorkflowDefinition( WorkflowNodeDto node, IReadOnlyDictionary workflowLibrary) @@ -163,4 +215,21 @@ internal sealed class WorkflowRunner throw new InvalidOperationException( $"Sub-workflow node \"{node.Id}\" references unknown workflow \"{node.Config.WorkflowId}\"."); } + + private static string NormalizeRequired(string? value, string errorMessage) + => NormalizeOptionalString(value) ?? throw new InvalidOperationException(errorMessage); + + private static string? NormalizeOptionalString(string? value) + => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); + + private sealed record WorkflowNodeRoute( + ExecutorBinding Entry, + ExecutorBinding Exit, + IReadOnlyList<(ExecutorBinding Source, ExecutorBinding Target)> InternalEdges) + { + public WorkflowNodeRoute(ExecutorBinding binding) + : this(binding, binding, []) + { + } + } } diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowValidator.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowValidator.cs index 491449e..11cebaa 100644 --- a/sidecar/src/Aryx.AgentHost/Services/WorkflowValidator.cs +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowValidator.cs @@ -10,7 +10,10 @@ public sealed class WorkflowValidator "start", "end", "agent", + "code-executor", + "function-executor", "sub-workflow", + "request-port", }; public IReadOnlyList Validate( @@ -103,6 +106,7 @@ public sealed class WorkflowValidator } } + ValidateExecutableNode(node, issues); ValidateSubWorkflowNode(node, workflowLibraryById, issues); } @@ -158,6 +162,9 @@ public sealed class WorkflowValidator List executableWorkNodes = workflow.Graph.Nodes .Where(node => string.Equals(node.Kind, "agent", StringComparison.OrdinalIgnoreCase) + || string.Equals(node.Kind, "code-executor", StringComparison.OrdinalIgnoreCase) + || string.Equals(node.Kind, "function-executor", StringComparison.OrdinalIgnoreCase) + || string.Equals(node.Kind, "request-port", StringComparison.OrdinalIgnoreCase) || string.Equals(node.Kind, "sub-workflow", StringComparison.OrdinalIgnoreCase)) .ToList(); @@ -184,7 +191,7 @@ public sealed class WorkflowValidator issues.Add(new WorkflowValidationIssueDto { Field = "graph.nodes", - Message = "Workflow graphs must contain at least one agent or sub-workflow node.", + Message = "Workflow graphs must contain at least one executable work node.", }); } @@ -405,6 +412,76 @@ public sealed class WorkflowValidator } } + private static void ValidateExecutableNode( + WorkflowNodeDto node, + List issues) + { + if (string.Equals(node.Kind, "code-executor", StringComparison.OrdinalIgnoreCase)) + { + if (string.IsNullOrWhiteSpace(node.Config.Implementation)) + { + issues.Add(new WorkflowValidationIssueDto + { + Field = "graph.nodes.config.implementation", + NodeId = node.Id, + Message = "Code executor nodes require a non-empty implementation.", + }); + } + + return; + } + + if (string.Equals(node.Kind, "function-executor", StringComparison.OrdinalIgnoreCase)) + { + if (string.IsNullOrWhiteSpace(node.Config.FunctionRef)) + { + issues.Add(new WorkflowValidationIssueDto + { + Field = "graph.nodes.config.functionRef", + NodeId = node.Id, + Message = "Function executor nodes require a non-empty functionRef.", + }); + } + + return; + } + + if (!string.Equals(node.Kind, "request-port", StringComparison.OrdinalIgnoreCase)) + { + return; + } + + if (string.IsNullOrWhiteSpace(node.Config.PortId)) + { + issues.Add(new WorkflowValidationIssueDto + { + Field = "graph.nodes.config.portId", + NodeId = node.Id, + Message = "Request port nodes require a non-empty portId.", + }); + } + + if (string.IsNullOrWhiteSpace(node.Config.RequestType)) + { + issues.Add(new WorkflowValidationIssueDto + { + Field = "graph.nodes.config.requestType", + NodeId = node.Id, + Message = "Request port nodes require a non-empty requestType.", + }); + } + + if (string.IsNullOrWhiteSpace(node.Config.ResponseType)) + { + issues.Add(new WorkflowValidationIssueDto + { + Field = "graph.nodes.config.responseType", + NodeId = node.Id, + Message = "Request port nodes require a non-empty responseType.", + }); + } + } + private static void ValidateEdgeCondition(WorkflowEdgeDto edge, List issues) { if (edge.Condition is null) diff --git a/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs index 958d3be..371114e 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs @@ -1,11 +1,13 @@ using System.IO; using System.Reflection; using System.Runtime.CompilerServices; +using System.Text.Json; using Aryx.AgentHost.Contracts; using Aryx.AgentHost.Services; using GitHub.Copilot.SDK; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Workflows; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Extensions.AI; namespace Aryx.AgentHost.Tests; @@ -799,6 +801,73 @@ public sealed class CopilotWorkflowRunnerTests }); } + [Fact] + public void CreateExecutionEnvironment_UsesLockstepWhenRequested() + { + RunTurnCommandDto command = new() + { + Workflow = new WorkflowDefinitionDto + { + Settings = new WorkflowSettingsDto + { + ExecutionMode = "lockstep", + Checkpointing = new WorkflowCheckpointSettingsDto(), + }, + }, + }; + + InProcessExecutionEnvironment environment = CopilotWorkflowRunner.CreateExecutionEnvironment(command, checkpointManager: null); + + Assert.Same(InProcessExecution.Lockstep, environment); + } + + [Fact] + public void CoerceRequestPortResponse_ParsesSupportedResponseTypes() + { + Assert.Equal("hello", CopilotWorkflowRunner.CoerceRequestPortResponse("string", "hello")); + Assert.Equal(true, CopilotWorkflowRunner.CoerceRequestPortResponse("bool", "yes")); + Assert.Equal(12.5d, CopilotWorkflowRunner.CoerceRequestPortResponse("number", "12.5")); + + JsonElement json = Assert.IsType(CopilotWorkflowRunner.CoerceRequestPortResponse("json", "{\"ok\":true}")); + Assert.True(json.GetProperty("ok").GetBoolean()); + } + + [Fact] + public async Task RunTurnAsync_RequestPortWorkflowUsesUserInputBridge() + { + CopilotWorkflowRunner runner = new(new PatternValidator()); + List requests = []; + + IReadOnlyList messages = await runner.RunTurnAsync( + CreateRequestPortCommand(), + _ => Task.CompletedTask, + _ => Task.CompletedTask, + _ => Task.CompletedTask, + async request => + { + requests.Add(request); + await runner.ResolveUserInputAsync( + new ResolveUserInputCommandDto + { + UserInputId = request.UserInputId, + Answer = "approved", + WasFreeform = true, + }, + CancellationToken.None); + }, + _ => Task.CompletedTask, + _ => Task.CompletedTask, + CancellationToken.None); + + UserInputRequestedEventDto requestEvent = Assert.Single(requests); + Assert.Equal("Needs approval?", requestEvent.Question); + Assert.Null(requestEvent.AgentId); + + ChatMessageDto message = Assert.Single(messages); + Assert.Equal("Workflow", message.AuthorName); + Assert.Equal("approved", message.Content); + } + [Fact] public async Task HandleWorkflowEventAsync_FallsBackToActiveAgentForUnresolvedStreamingUpdates() { @@ -2212,6 +2281,96 @@ public sealed class CopilotWorkflowRunnerTests }; } + private static RunTurnCommandDto CreateRequestPortCommand() + { + return new RunTurnCommandDto + { + RequestId = "turn-request-port", + SessionId = "session-request-port", + ProjectPath = "c:\\workspace\\personal\\projects\\aryx.worktrees\\copilot-powerful-vulture", + Pattern = new PatternDefinitionDto + { + Id = "pattern-request-port", + Name = "Request Port Pattern", + Mode = "single", + Availability = "available", + Agents = [], + }, + Messages = + [ + new ChatMessageDto + { + Id = "message-1", + Role = "user", + AuthorName = "User", + Content = "Please continue.", + CreatedAt = "2026-04-05T00:00:00.000Z", + }, + ], + Workflow = new WorkflowDefinitionDto + { + Id = "workflow-request-port", + Name = "Request Port Workflow", + Graph = new WorkflowGraphDto + { + Nodes = + [ + new WorkflowNodeDto + { + Id = "start", + Kind = "start", + Label = "Start", + Config = new WorkflowNodeConfigDto { Kind = "start" }, + }, + new WorkflowNodeDto + { + Id = "approval-port", + Kind = "request-port", + Label = "Approval", + Config = new WorkflowNodeConfigDto + { + Kind = "request-port", + PortId = "approval", + RequestType = "Question", + ResponseType = "string", + Prompt = "Needs approval?", + }, + }, + new WorkflowNodeDto + { + Id = "end", + Kind = "end", + Label = "End", + Config = new WorkflowNodeConfigDto { Kind = "end" }, + }, + ], + Edges = + [ + new WorkflowEdgeDto + { + Id = "edge-start-port", + Source = "start", + Target = "approval-port", + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-port-end", + Source = "approval-port", + Target = "end", + Kind = "direct", + }, + ], + }, + Settings = new WorkflowSettingsDto + { + Checkpointing = new WorkflowCheckpointSettingsDto(), + ExecutionMode = "lockstep", + }, + }, + }; + } + private static RunTurnMcpServerConfigDto CreateMcpServerConfig(string serverName) => new() { diff --git a/sidecar/tests/Aryx.AgentHost.Tests/SidecarProtocolHostTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/SidecarProtocolHostTests.cs index a1d2be1..b3a10c6 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/SidecarProtocolHostTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/SidecarProtocolHostTests.cs @@ -169,7 +169,7 @@ public sealed class SidecarProtocolHostTests && issue.GetProperty("message").GetString() == "Workflow name is required."); Assert.Contains(issues, issue => issue.GetProperty("field").GetString() == "graph.nodes" - && issue.GetProperty("message").GetString() == "Workflow graphs must contain at least one agent or sub-workflow node."); + && issue.GetProperty("message").GetString() == "Workflow graphs must contain at least one executable work node."); }, completionEvent => { diff --git a/sidecar/tests/Aryx.AgentHost.Tests/WorkflowRunnerTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/WorkflowRunnerTests.cs index a332083..c26d080 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/WorkflowRunnerTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/WorkflowRunnerTests.cs @@ -3,6 +3,7 @@ using Aryx.AgentHost.Services; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Workflows; using Microsoft.Extensions.AI; +using System.Text.Json; namespace Aryx.AgentHost.Tests; @@ -52,6 +53,112 @@ public sealed class WorkflowRunnerTests Assert.Contains("unknown workflow", error.Message, StringComparison.OrdinalIgnoreCase); } + [Fact] + public async Task BuildWorkflow_RunsCodeExecutorAndSurfacesOutput() + { + WorkflowRunner runner = new(); + Workflow workflow = runner.BuildWorkflow( + CreateSingleNodeWorkflow( + "code-executor", + new WorkflowNodeConfigDto + { + Kind = "code-executor", + Implementation = "return-text:done", + }), + CreateEmptyPattern(), + []); + + List output = await RunWorkflowToOutputAsync(workflow); + + ChatMessage message = Assert.Single(output); + Assert.Equal("done", message.Text); + Assert.Equal("Workflow", message.AuthorName); + } + + [Fact] + public async Task BuildWorkflow_FunctionExecutorsUseStateScopes() + { + WorkflowRunner runner = new(); + Workflow workflow = runner.BuildWorkflow( + CreateStatefulFunctionWorkflow(), + CreateEmptyPattern(), + []); + + List output = await RunWorkflowToOutputAsync(workflow); + + ChatMessage message = Assert.Single(output); + Assert.Equal("{\"status\":\"complete\"}", message.Text); + } + + [Fact] + public async Task BuildWorkflow_RequestPortsRaiseRequestsAndForwardResponses() + { + WorkflowRunner runner = new(); + Workflow workflow = runner.BuildWorkflow( + CreateSingleNodeWorkflow( + "request-port", + new WorkflowNodeConfigDto + { + Kind = "request-port", + PortId = "approval", + RequestType = "Question", + ResponseType = "string", + Prompt = "Approve the workflow?", + }), + CreateEmptyPattern(), + []); + + ChatMessage[] input = + [ + new(ChatRole.User, "Please continue."), + ]; + + await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, input); + + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is RequestInfoEvent requestInfo) + { + Assert.Equal("approval", requestInfo.Request.PortInfo.PortId); + WorkflowRequestPortPromptRequest payload = Assert.IsType( + requestInfo.Request.Data.As()); + Assert.Equal("Approve the workflow?", payload.Prompt); + Assert.Equal("request-port", payload.NodeId); + await run.SendResponseAsync(requestInfo.Request.CreateResponse("approved")); + continue; + } + + if (evt is WorkflowOutputEvent outputEvent) + { + List output = Assert.IsType>(outputEvent.Data); + ChatMessage message = Assert.Single(output); + Assert.Equal("approved", message.Text); + return; + } + } + + Assert.Fail("Workflow never produced an output after the request port response."); + } + + [Fact] + public void BuildWorkflow_RejectsUnknownFunctionRefsAtBuildTime() + { + WorkflowRunner runner = new(); + + InvalidOperationException error = Assert.Throws(() => runner.BuildWorkflow( + CreateSingleNodeWorkflow( + "function-executor", + new WorkflowNodeConfigDto + { + Kind = "function-executor", + FunctionRef = "missing-function", + }), + CreateEmptyPattern(), + [])); + + Assert.Contains("unsupported functionRef", error.Message, StringComparison.OrdinalIgnoreCase); + } + private static PatternDefinitionDto CreatePattern(string agentId) { return new PatternDefinitionDto @@ -136,6 +243,18 @@ public sealed class WorkflowRunnerTests }; } + private static PatternDefinitionDto CreateEmptyPattern() + { + return new PatternDefinitionDto + { + Id = "pattern-empty", + Name = "Workflow Pattern", + Mode = "single", + Availability = "available", + Agents = [], + }; + } + private static WorkflowDefinitionDto CreateSubworkflowParent( string? workflowId = null, WorkflowDefinitionDto? inlineWorkflow = null) @@ -200,6 +319,203 @@ public sealed class WorkflowRunnerTests }; } + private static WorkflowDefinitionDto CreateSingleNodeWorkflow(string nodeKind, WorkflowNodeConfigDto config) + { + return new WorkflowDefinitionDto + { + Id = $"workflow-{nodeKind}", + Name = $"{nodeKind} Workflow", + Graph = new WorkflowGraphDto + { + Nodes = + [ + new WorkflowNodeDto + { + Id = "start", + Kind = "start", + Label = "Start", + Config = new WorkflowNodeConfigDto { Kind = "start" }, + }, + new WorkflowNodeDto + { + Id = nodeKind, + Kind = nodeKind, + Label = nodeKind, + Config = config, + }, + new WorkflowNodeDto + { + Id = "end", + Kind = "end", + Label = "End", + Config = new WorkflowNodeConfigDto { Kind = "end" }, + }, + ], + Edges = + [ + new WorkflowEdgeDto + { + Id = "edge-start-node", + Source = "start", + Target = nodeKind, + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-node-end", + Source = nodeKind, + Target = "end", + Kind = "direct", + }, + ], + }, + Settings = new WorkflowSettingsDto + { + Checkpointing = new WorkflowCheckpointSettingsDto(), + }, + }; + } + + private static WorkflowDefinitionDto CreateStatefulFunctionWorkflow() + { + return new WorkflowDefinitionDto + { + Id = "workflow-stateful-function", + Name = "Stateful Function Workflow", + Graph = new WorkflowGraphDto + { + Nodes = + [ + new WorkflowNodeDto + { + Id = "start", + Kind = "start", + Label = "Start", + Config = new WorkflowNodeConfigDto { Kind = "start" }, + }, + new WorkflowNodeDto + { + Id = "state-get", + Kind = "function-executor", + Label = "Get State", + Config = new WorkflowNodeConfigDto + { + Kind = "function-executor", + FunctionRef = "state:get", + Parameters = new Dictionary + { + ["scope"] = JsonDocument.Parse("\"workflow\"").RootElement.Clone(), + ["key"] = JsonDocument.Parse("\"status\"").RootElement.Clone(), + }, + }, + }, + new WorkflowNodeDto + { + Id = "state-set", + Kind = "function-executor", + Label = "Set State", + Config = new WorkflowNodeConfigDto + { + Kind = "function-executor", + FunctionRef = "state:set", + Parameters = new Dictionary + { + ["scope"] = JsonDocument.Parse("\"workflow\"").RootElement.Clone(), + ["key"] = JsonDocument.Parse("\"status\"").RootElement.Clone(), + ["value"] = JsonDocument.Parse("{\"status\":\"complete\"}").RootElement.Clone(), + }, + }, + }, + new WorkflowNodeDto + { + Id = "state-read-back", + Kind = "code-executor", + Label = "Read Back", + Config = new WorkflowNodeConfigDto + { + Kind = "code-executor", + Implementation = "state:get:workflow:status", + }, + }, + new WorkflowNodeDto + { + Id = "end", + Kind = "end", + Label = "End", + Config = new WorkflowNodeConfigDto { Kind = "end" }, + }, + ], + Edges = + [ + new WorkflowEdgeDto + { + Id = "edge-start-get", + Source = "start", + Target = "state-get", + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-get-set", + Source = "state-get", + Target = "state-set", + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-set-read", + Source = "state-set", + Target = "state-read-back", + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-read-end", + Source = "state-read-back", + Target = "end", + Kind = "direct", + }, + ], + }, + Settings = new WorkflowSettingsDto + { + Checkpointing = new WorkflowCheckpointSettingsDto(), + StateScopes = + [ + new WorkflowStateScopeDto + { + Name = "workflow", + InitialValues = new Dictionary + { + ["status"] = JsonDocument.Parse("\"pending\"").RootElement.Clone(), + }, + }, + ], + }, + }; + } + + private static async Task> RunWorkflowToOutputAsync(Workflow workflow) + { + ChatMessage[] input = + [ + new(ChatRole.User, "Run the workflow."), + ]; + + await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, input); + + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is WorkflowOutputEvent outputEvent) + { + return Assert.IsType>(outputEvent.Data); + } + } + + Assert.Fail("Workflow did not produce an output."); + return []; + } + private static ChatClientAgent CreateChatClientAgent(string id, string name) { return new ChatClientAgent( diff --git a/sidecar/tests/Aryx.AgentHost.Tests/WorkflowValidatorTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/WorkflowValidatorTests.cs index 7e2d9a1..d180f99 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/WorkflowValidatorTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/WorkflowValidatorTests.cs @@ -152,6 +152,74 @@ public sealed class WorkflowValidatorTests Assert.DoesNotContain(issues, issue => issue.Level == "error"); } + [Fact] + public void Validate_AcceptsPhase4ExecutableNodeKinds() + { + WorkflowDefinitionDto codeWorkflow = CreateSingleNodeWorkflow( + "code-executor", + new WorkflowNodeConfigDto + { + Kind = "code-executor", + Implementation = "return-text:done", + }); + WorkflowDefinitionDto functionWorkflow = CreateSingleNodeWorkflow( + "function-executor", + new WorkflowNodeConfigDto + { + Kind = "function-executor", + FunctionRef = "identity", + }); + WorkflowDefinitionDto requestPortWorkflow = CreateSingleNodeWorkflow( + "request-port", + new WorkflowNodeConfigDto + { + Kind = "request-port", + PortId = "approval", + RequestType = "Question", + ResponseType = "string", + }); + + Assert.DoesNotContain(_validator.Validate(codeWorkflow), issue => issue.Level == "error"); + Assert.DoesNotContain(_validator.Validate(functionWorkflow), issue => issue.Level == "error"); + Assert.DoesNotContain(_validator.Validate(requestPortWorkflow), issue => issue.Level == "error"); + } + + [Fact] + public void Validate_RejectsInvalidPhase4ExecutorConfigs() + { + WorkflowDefinitionDto codeWorkflow = CreateSingleNodeWorkflow( + "code-executor", + new WorkflowNodeConfigDto + { + Kind = "code-executor", + Implementation = " ", + }); + WorkflowDefinitionDto functionWorkflow = CreateSingleNodeWorkflow( + "function-executor", + new WorkflowNodeConfigDto + { + Kind = "function-executor", + FunctionRef = string.Empty, + }); + WorkflowDefinitionDto requestPortWorkflow = CreateSingleNodeWorkflow( + "request-port", + new WorkflowNodeConfigDto + { + Kind = "request-port", + PortId = " ", + RequestType = "", + ResponseType = null, + }); + + Assert.Contains(_validator.Validate(codeWorkflow), issue => issue.Field == "graph.nodes.config.implementation"); + Assert.Contains(_validator.Validate(functionWorkflow), issue => issue.Field == "graph.nodes.config.functionRef"); + + IReadOnlyList requestPortIssues = _validator.Validate(requestPortWorkflow); + Assert.Contains(requestPortIssues, issue => issue.Field == "graph.nodes.config.portId"); + Assert.Contains(requestPortIssues, issue => issue.Field == "graph.nodes.config.requestType"); + Assert.Contains(requestPortIssues, issue => issue.Field == "graph.nodes.config.responseType"); + } + private static WorkflowDefinitionDto CreateWorkflow(string id = "workflow-1") { return new WorkflowDefinitionDto @@ -278,4 +346,61 @@ public sealed class WorkflowValidatorTests }, }; } + + private static WorkflowDefinitionDto CreateSingleNodeWorkflow(string nodeKind, WorkflowNodeConfigDto config) + { + return new WorkflowDefinitionDto + { + Id = $"workflow-{nodeKind}", + Name = $"{nodeKind} Workflow", + Graph = new WorkflowGraphDto + { + Nodes = + [ + new WorkflowNodeDto + { + Id = "start", + Kind = "start", + Label = "Start", + Config = new WorkflowNodeConfigDto { Kind = "start" }, + }, + new WorkflowNodeDto + { + Id = nodeKind, + Kind = nodeKind, + Label = nodeKind, + Config = config, + }, + new WorkflowNodeDto + { + Id = "end", + Kind = "end", + Label = "End", + Config = new WorkflowNodeConfigDto { Kind = "end" }, + }, + ], + Edges = + [ + new WorkflowEdgeDto + { + Id = "edge-start-node", + Source = "start", + Target = nodeKind, + Kind = "direct", + }, + new WorkflowEdgeDto + { + Id = "edge-node-end", + Source = nodeKind, + Target = "end", + Kind = "direct", + }, + ], + }, + Settings = new WorkflowSettingsDto + { + Checkpointing = new WorkflowCheckpointSettingsDto(), + }, + }; + } } diff --git a/src/shared/domain/workflow.ts b/src/shared/domain/workflow.ts index 331fabb..b3870ed 100644 --- a/src/shared/domain/workflow.ts +++ b/src/shared/domain/workflow.ts @@ -170,7 +170,15 @@ export interface WorkflowResolutionOptions { resolveWorkflow?: (workflowId: string) => WorkflowDefinition | undefined; } -const executableNodeKinds = new Set(['start', 'end', 'agent', 'sub-workflow']); +const executableNodeKinds = new Set([ + 'start', + 'end', + 'agent', + 'code-executor', + 'function-executor', + 'sub-workflow', + 'request-port', +]); function normalizeOptionalString(value?: string): string | undefined { const trimmed = value?.trim(); @@ -740,6 +748,66 @@ function validateSubWorkflowNode(node: WorkflowNode, issues: WorkflowValidationI } } +function validateExecutableNodeConfig(node: WorkflowNode, issues: WorkflowValidationIssue[]): void { + switch (node.kind) { + case 'code-executor': + if (node.config.kind === 'code-executor' && !node.config.implementation?.trim()) { + addIssue(issues, { + level: 'error', + field: 'graph.nodes.config.implementation', + nodeId: node.id, + message: 'Code executor nodes require a non-empty implementation.', + }); + } + return; + case 'function-executor': + if (node.config.kind === 'function-executor' && !node.config.functionRef.trim()) { + addIssue(issues, { + level: 'error', + field: 'graph.nodes.config.functionRef', + nodeId: node.id, + message: 'Function executor nodes require a non-empty functionRef.', + }); + } + return; + case 'request-port': + if (node.config.kind !== 'request-port') { + return; + } + + if (!node.config.portId.trim()) { + addIssue(issues, { + level: 'error', + field: 'graph.nodes.config.portId', + nodeId: node.id, + message: 'Request port nodes require a non-empty portId.', + }); + } + + if (!node.config.requestType.trim()) { + addIssue(issues, { + level: 'error', + field: 'graph.nodes.config.requestType', + nodeId: node.id, + message: 'Request port nodes require a non-empty requestType.', + }); + } + + if (!node.config.responseType.trim()) { + addIssue(issues, { + level: 'error', + field: 'graph.nodes.config.responseType', + nodeId: node.id, + message: 'Request port nodes require a non-empty responseType.', + }); + } + + return; + default: + return; + } +} + export function validateWorkflowDefinition(workflow: WorkflowDefinition): WorkflowValidationIssue[] { const normalized = normalizeWorkflowDefinition(workflow); const issues: WorkflowValidationIssue[] = []; @@ -818,6 +886,7 @@ export function validateWorkflowDefinition(workflow: WorkflowDefinition): Workfl } } + validateExecutableNodeConfig(node, issues); validateSubWorkflowNode(node, issues); } for (const edge of normalized.graph.edges) { @@ -860,7 +929,11 @@ export function validateWorkflowDefinition(workflow: WorkflowDefinition): Workfl const startNodes = normalized.graph.nodes.filter((node) => node.kind === 'start'); const endNodes = normalized.graph.nodes.filter((node) => node.kind === 'end'); const executableWorkNodes = normalized.graph.nodes.filter((node) => - node.kind === 'agent' || node.kind === 'sub-workflow'); + node.kind === 'agent' + || node.kind === 'code-executor' + || node.kind === 'function-executor' + || node.kind === 'sub-workflow' + || node.kind === 'request-port'); if (startNodes.length !== 1) { addIssue(issues, { @@ -882,7 +955,7 @@ export function validateWorkflowDefinition(workflow: WorkflowDefinition): Workfl addIssue(issues, { level: 'error', field: 'graph.nodes', - message: 'Workflow graphs must contain at least one agent or sub-workflow node.', + message: 'Workflow graphs must contain at least one executable work node.', }); } diff --git a/tests/shared/workflow.test.ts b/tests/shared/workflow.test.ts index 20b8272..883624d 100644 --- a/tests/shared/workflow.test.ts +++ b/tests/shared/workflow.test.ts @@ -116,23 +116,111 @@ describe('workflow validation', () => { expect(validateWorkflowDefinition(createWorkflow())).toEqual([]); }); - test('rejects unsupported node kinds in phase 1 backend', () => { + test('accepts phase 4 executor node kinds as executable work', () => { const workflow = createWorkflow(); - workflow.graph.nodes.splice(1, 0, { + workflow.graph.nodes[1] = { + id: 'code-executor', + kind: 'code-executor', + label: 'Transform', + position: { x: 100, y: 0 }, + config: { + kind: 'code-executor', + implementation: 'return-text:done', + }, + }; + workflow.graph.edges[0] = { id: 'edge-start-code', source: 'start', target: 'code-executor', kind: 'direct' }; + workflow.graph.edges[1] = { id: 'edge-code-end', source: 'code-executor', target: 'end', kind: 'direct' }; + + expect(validateWorkflowDefinition(workflow)).toEqual([]); + }); + + test('counts function and request port nodes as executable work', () => { + const functionWorkflow = createWorkflow(); + functionWorkflow.graph.nodes[1] = { + id: 'function-executor', + kind: 'function-executor', + label: 'Function', + position: { x: 200, y: 0 }, + config: { + kind: 'function-executor', + functionRef: 'identity', + }, + }; + functionWorkflow.graph.edges[0] = { id: 'edge-start-function', source: 'start', target: 'function-executor', kind: 'direct' }; + functionWorkflow.graph.edges[1] = { id: 'edge-function-end', source: 'function-executor', target: 'end', kind: 'direct' }; + + const requestPortWorkflow = createWorkflow(); + requestPortWorkflow.graph.nodes[1] = { id: 'request-port', kind: 'request-port', label: 'Approval', - position: { x: 100, y: 0 }, + position: { x: 200, y: 0 }, config: { kind: 'request-port', portId: 'approval', requestType: 'Question', - responseType: 'Answer', + responseType: 'string', }, - }); + }; + requestPortWorkflow.graph.edges[0] = { id: 'edge-start-port', source: 'start', target: 'request-port', kind: 'direct' }; + requestPortWorkflow.graph.edges[1] = { id: 'edge-port-end', source: 'request-port', target: 'end', kind: 'direct' }; - expect(validateWorkflowDefinition(workflow).some((issue) => - issue.message.includes('not executable yet'))).toBe(true); + expect(validateWorkflowDefinition(functionWorkflow)).toEqual([]); + expect(validateWorkflowDefinition(requestPortWorkflow)).toEqual([]); + }); + + test('rejects invalid phase 4 executor configs', () => { + const codeWorkflow = createWorkflow(); + codeWorkflow.graph.nodes[1] = { + id: 'code-executor', + kind: 'code-executor', + label: 'Code', + position: { x: 200, y: 0 }, + config: { + kind: 'code-executor', + implementation: ' ', + }, + }; + codeWorkflow.graph.edges[0] = { id: 'edge-start-code', source: 'start', target: 'code-executor', kind: 'direct' }; + codeWorkflow.graph.edges[1] = { id: 'edge-code-end', source: 'code-executor', target: 'end', kind: 'direct' }; + + const functionWorkflow = createWorkflow(); + functionWorkflow.graph.nodes[1] = { + id: 'function-executor', + kind: 'function-executor', + label: 'Function', + position: { x: 200, y: 0 }, + config: { + kind: 'function-executor', + functionRef: ' ', + }, + }; + functionWorkflow.graph.edges[0] = { id: 'edge-start-function', source: 'start', target: 'function-executor', kind: 'direct' }; + functionWorkflow.graph.edges[1] = { id: 'edge-function-end', source: 'function-executor', target: 'end', kind: 'direct' }; + + const requestPortWorkflow = createWorkflow(); + requestPortWorkflow.graph.nodes[1] = { + id: 'request-port', + kind: 'request-port', + label: 'Approval', + position: { x: 200, y: 0 }, + config: { + kind: 'request-port', + portId: ' ', + requestType: '', + responseType: ' ', + }, + }; + requestPortWorkflow.graph.edges[0] = { id: 'edge-start-port', source: 'start', target: 'request-port', kind: 'direct' }; + requestPortWorkflow.graph.edges[1] = { id: 'edge-port-end', source: 'request-port', target: 'end', kind: 'direct' }; + + expect(validateWorkflowDefinition(codeWorkflow).some((issue) => issue.field === 'graph.nodes.config.implementation')).toBe(true); + expect(validateWorkflowDefinition(functionWorkflow).some((issue) => issue.field === 'graph.nodes.config.functionRef')).toBe(true); + + const requestPortIssues = validateWorkflowDefinition(requestPortWorkflow); + expect(requestPortIssues.some((issue) => issue.field === 'graph.nodes.config.portId')).toBe(true); + expect(requestPortIssues.some((issue) => issue.field === 'graph.nodes.config.requestType')).toBe(true); + expect(requestPortIssues.some((issue) => issue.field === 'graph.nodes.config.responseType')).toBe(true); }); test('accepts sub-workflow nodes with inline workflows', () => {