XFEWorkflow

using System.IO.Pipes;
using System.Text.Json;
using System.Text.Json.Nodes;
using XFEWorkflow.Abstractions;
using XFEWorkflow.Core.Runtime;
using XFEWorkflow.Infrastructure.Communication;
using XFEWorkflow.Infrastructure.Compilation;
using XFEWorkflow.Infrastructure.Plugins;
using XFEWorkflow.Infrastructure.Storage;
using XFEWorkflow.Infrastructure.Execution;

if (args.Length == 2 && args[0] == "--command-host") return await CommandProcessWorker.RunAsync(args[1]);

if (args.Length != 2 || args[0] != "--pipe") { Console.Error.WriteLine("Runner 由 XFEWorkflow 启动。"); return 2; }
using var lifetime = new CancellationTokenSource();
await using var pipe = new NamedPipeClientStream(".", args[1], PipeDirection.InOut, PipeOptions.Asynchronous);
try { await pipe.ConnectAsync(20_000, lifetime.Token); } catch (Exception ex) { Console.Error.WriteLine(ex.Message); return 3; }
await using var peer = new RpcPeer(pipe);
CancellationTokenSource? execution = null;
WorkflowDebugger? debugger = null;
var running = 0;
peer.RequestReceived = async (method, value, ct) =>
{
    if (method == "cancel") { if (execution is not null) await execution.CancelAsync(); return new JsonObject { ["cancelled"] = true }; }
    if (method == "debug") return new JsonObject { ["accepted"] = debugger?.Command(value!.Deserialize<WorkflowDebugCommand>(WorkflowJson.Options)!) == true };
    if (method != "run") throw new NotSupportedException(method);
    if (Interlocked.Exchange(ref running, 1) != 0) throw new InvalidOperationException("Runner 只接受一个运行任务");
    var request = value!.Deserialize<RunRequest>(WorkflowJson.Options)!;
    execution = CancellationTokenSource.CreateLinkedTokenSource(ct, lifetime.Token);
    var token = execution.Token;
    var response = new RunResponse();
    try
    {
        debugger = request.Debug is { } debugSettings ? new(debugSettings, (state, cancellation) => peer.EventAsync("run.debug", state, cancellation)) : null;
        using var loader = new PluginLoader(); loader.Load(request.Plugins);
        using var compiler = new CodeCompiler { CapabilityBindingsSource = PluginMethodBindings.Generate(loader.Registry.Descriptors.Values) };
        var engine = new WorkflowEngine(compiler, CoreCapabilities.Catalog(loader.Registry.Descriptors.Values));
        var repository = new WorkflowRepository(new WorkspacePaths(request.WorkspaceRoot));
        var depth = new AsyncLocal<int>();
        async ValueTask<JsonObject> Execute(WorkflowDefinition workflow, JsonObject inputs, CancellationToken cancellation, int debugDepth = 0)
        {
            if (depth.Value >= 32) throw new InvalidOperationException("子工作流调用层数超过 32");
            depth.Value++;
            try
            {
                foreach (var locked in workflow.PluginVersions)
                    if (!request.Plugins.Any(x => x.Manifest.Id == locked.Key && x.Manifest.Version == locked.Value))
                        throw new InvalidOperationException("当前运行未加载子工作流锁定的插件：" + locked.Key);
                var context = CreateContext(workflow, inputs, depth.Value == 1 ? request.EntryNodeId : null, debugDepth);
                return await engine.ExecuteAsync(workflow, context, cancellation);
            }
            finally { depth.Value--; }
        }
        WorkflowExecutionContext CreateContext(WorkflowDefinition workflow, JsonObject inputs, string? entryNodeId = null, int debugDepth = 0) => new()
        {
            RunId = request.RunId, Inputs = WorkflowJson.Copy(inputs), Variables = WorkflowJson.Copy(workflow.Variables), Capabilities = loader.Registry, EntryNodeId = entryNodeId,
            WorkflowId = workflow.Id, Debugger = debugger, DebugDepth = debugDepth,
            Subworkflow = async (id, data, cancellation) => await Execute(await repository.GetAsync(id, cancellation), data, cancellation),
            SubworkflowWithContext = async (id, data, parent, cancellation) => await Execute(await repository.GetAsync(id, cancellation), data, cancellation, parent.DebugDepth + 1),
            Secret = async (reference, cancellation) => await peer.RequestAsync<string>("secret", new { reference }, cancellation) ?? "",
            Host = async (operation, data, cancellation) => await peer.RequestAsync<JsonObject>("host", new { method = operation, parameters = data }, cancellation) ?? [],
            Report = ev => peer.EventAsync("run.event", ev, token)
        };
        if (request.TriggerCapability is { } trigger)
        {
            await loader.Registry.RunTriggerAsync(trigger, request.TriggerParameters, CreateContext(request.Workflow, request.Inputs),
                data => peer.EventAsync("trigger.fire", data, token), token);
            response.Status = RunStatus.Succeeded;
        }
        else
        {
            await peer.EventAsync("run.event", new RunEvent { RunId = request.RunId, Kind = RunEventKind.Started, Message = "工作流开始" }, token);
            response.Outputs = await Execute(request.Workflow, request.Inputs, token);
            response.Status = RunStatus.Succeeded;
        }
    }
    catch (OperationCanceledException) when (token.IsCancellationRequested) { response.Status = RunStatus.Cancelled; response.Error = "运行已停止"; }
    catch (Exception ex) { response.Status = RunStatus.Failed; response.Error = ex.Message; response.NodeId = (ex as WorkflowExecutionException)?.NodeId; }
    finally { debugger = null; execution.Dispose(); execution = null; }
    return JsonSerializer.SerializeToNode(response, WorkflowJson.Options);
};
peer.Start();
await peer.Completion;
await lifetime.CancelAsync();
return 0;