返回提交历史
Modified
XFEWorkflow/XFEWorkflow.Abstractions/PluginContracts.cs
+8
-1
Modified
XFEWorkflow/XFEWorkflow.Abstractions/PluginPackage.cs
+2
-1
Modified
XFEWorkflow/XFEWorkflow.Core/Runtime/GraphValidator.cs
+5
-1
Added
XFEWorkflow/XFEWorkflow.Core/Runtime/NodeEncapsulation.cs
+87
-0
Modified
XFEWorkflow/XFEWorkflow.Core/Runtime/WorkflowEngine.cs
+22
-6
Modified
XFEWorkflow/XFEWorkflow.Core/Runtime/WorkflowExecutionContext.cs
+11
-1
Added
XFEWorkflow/XFEWorkflow.Core/Runtime/WorkflowPortCatalog.cs
+17
-0
Modified
XFEWorkflow/XFEWorkflow.Infrastructure/Compilation/CodeCompiler.cs
+13
-2
Added
XFEWorkflow/XFEWorkflow.Infrastructure/Compilation/PluginMethodBindings.cs
+42
-0
Added
XFEWorkflow/XFEWorkflow.Infrastructure/Plugins/GraphPluginBuilder.cs
+89
-0
Added
XFEWorkflow/XFEWorkflow.Infrastructure/Plugins/PluginDependencyResolver.cs
+36
-0
Modified
XFEWorkflow/XFEWorkflow.Infrastructure/Plugins/PluginManager.cs
+3
-4
Modified
XFEWorkflow/XFEWorkflow.Infrastructure/Space/SpaceClient.cs
+43
-2
Modified
XFEWorkflow/XFEWorkflow.Runner/Program.cs
+1
-1
Added
XFEWorkflow/XFEWorkflow.Test/GraphPluginTests.cs
+70
-0
Modified
XFEWorkflow/XFEWorkflow.Test/SpaceProtocolTests.cs
+8
-0
Modified
XFEWorkflow/XFEWorkflow/Controls/CodeEditor.cs
+2
-1
Modified
XFEWorkflow/XFEWorkflow/Controls/WorkflowCanvas.cs
+3
-0
Added
XFEWorkflow/XFEWorkflow/Views/MainPage.PluginAuthoring.cs
+59
-0
Modified
XFEWorkflow/XFEWorkflow/Views/MainPage.xaml
+1
-0
Modified
XFEWorkflow/XFEWorkflow/Views/PluginsPage.xaml.cs
+11
-1
Added
docs/graph-plugin-authoring.md
+25
-0
XFEstudio/XFEWorkflow
支持图形方法封装为插件节点、元组返回值和依赖自动安装
a64c07d
代码差异
22 个文件
+558
-21
@@ -4,7 +4,7 @@ namespace XFEWorkflow.Abstractions;
4
4
5
5
public static class WorkflowSdk
6
6
{
7
public const string CurrentVersion = "1.2.0";
7
public const string CurrentVersion = "1.3.0";
8
8
}
9
9
10
10
public interface IWorkflow
@@ -21,6 +21,8 @@ public interface IWorkflowContext
21
21
ValueTask<JsonObject> InvokeWorkflowAsync(string workflowId, JsonObject inputs, CancellationToken cancellationToken = default);
22
22
ValueTask<JsonObject> ExecuteGraphAsync(GraphDefinition graph, CancellationToken cancellationToken = default)
23
23
=> throw new NotSupportedException("当前宿主不支持从 C# 执行图形，请使用 SDK 1.2 或更高版本");
24
ValueTask<JsonObject> ExecuteWorkflowAsync(WorkflowDefinition workflow, JsonObject inputs, CancellationToken cancellationToken = default)
25
=> throw new NotSupportedException("封装节点需要 SDK 1.3 或更高版本");
24
26
ValueTask<string> GetSecretAsync(string reference, CancellationToken cancellationToken = default);
25
27
ValueTask<JsonObject> InvokeHostAsync(string method, JsonObject parameters, CancellationToken cancellationToken = default);
26
28
ValueTask LogAsync(string message, CancellationToken cancellationToken = default);
@@ -63,6 +65,7 @@ public sealed class CapabilityDescriptor
63
65
public List<PortDefinition> Inputs { get; set; } = [];
64
66
public List<PortDefinition> Outputs { get; set; } = [];
65
67
public List<string> ExecutionOutputs { get; set; } = ["next"];
68
public string? ExecutionResultPort { get; set; }
66
69
public List<string> ChildGraphs { get; set; } = [];
67
70
public override string ToString() => Name;
68
71
}
@@ -91,6 +94,10 @@ public sealed class PluginDependency
91
94
public string Id { get; set; } = "";
92
95
public string MinimumVersion { get; set; } = "1.0.0";
93
96
public string? MaximumVersionExclusive { get; set; }
97
public string? ExactVersion { get; set; }
98
public bool Accepts(string version) => (ExactVersion is null || ExactVersion == version)
99
&& SemanticVersion.Parse(version).CompareTo(SemanticVersion.Parse(MinimumVersion)) >= 0
100
&& (MaximumVersionExclusive is null || SemanticVersion.Parse(version).CompareTo(SemanticVersion.Parse(MaximumVersionExclusive)) < 0);
94
101
}
95
102
96
103
public sealed class InstalledPlugin
@@ -73,7 +73,8 @@ public static partial class PluginPackage
73
73
foreach (var dependency in manifest.Dependencies)
74
74
if (!IsIdentifier(dependency.Id) || dependency.Id == manifest.Id ||
75
75
!SemanticVersion.TryParse(dependency.MinimumVersion, out _) ||
76
(dependency.MaximumVersionExclusive is not null && !SemanticVersion.TryParse(dependency.MaximumVersionExclusive, out _)))
76
(dependency.MaximumVersionExclusive is not null && !SemanticVersion.TryParse(dependency.MaximumVersionExclusive, out _)) ||
77
(dependency.ExactVersion is not null && !SemanticVersion.TryParse(dependency.ExactVersion, out _)))
77
78
throw new InvalidDataException("插件依赖声明无效");
78
79
}
79
80
public static void ValidateRelativePath(string path)
@@ -9,6 +9,10 @@ public static class GraphValidator
9
9
public static List<WorkflowDiagnostic> Validate(WorkflowDefinition workflow, IReadOnlyDictionary<string, CapabilityDescriptor> catalog)
10
10
{
11
11
List<WorkflowDiagnostic> errors = [];
12
catalog = WorkflowPortCatalog.For(workflow, catalog);
13
foreach (var ports in new[] { workflow.Inputs, workflow.Outputs })
14
if (ports.Any(x => string.IsNullOrWhiteSpace(x.Name) || x.Name == "inputs") || ports.Select(x => x.Name).Distinct().Count() != ports.Count)
15
errors.Add(new("参数与返回值名称不能为空、重复或使用保留名称 inputs"));
12
16
if (workflow.SchemaVersion != 1) errors.Add(new("不支持的工作流格式版本"));
13
17
if (string.IsNullOrWhiteSpace(workflow.Name)) errors.Add(new("工作流名称不能为空"));
14
18
if (workflow.Kind == WorkflowKind.Graph) ValidateGraph(workflow.Graph, catalog, errors, []);
@@ -43,7 +47,7 @@ public static class GraphValidator
43
47
if (depth > 0 && WorkflowTriggerNodes.IsTrigger(node)) errors.Add(new("自动触发节点只能放在工作流主图", node.Id));
44
48
if (node.RetryCount is < 0 or > 100 || node.TimeoutSeconds < 0 || node.RetryDelayMilliseconds < 0)
45
49
errors.Add(new("超时或重试设置无效", node.Id));
46
foreach (var child in node.Children.Values) ValidateGraph(child, catalog, errors, ids, depth + 1);
50
foreach (var child in node.Children.Values) ValidateGraph(child, CoreCapabilities.Catalog(catalog.Values.Where(x => x.PluginId != "core")), errors, ids, depth + 1);
47
51
}
48
52
foreach (var edge in graph.Edges)
49
53
{
@@ -0,0 +1,87 @@
1
using System.Text.Json.Nodes;
2
using XFEWorkflow.Abstractions;
3
4
namespace XFEWorkflow.Core.Runtime;
5
6
public sealed class NodeEncapsulation
7
{
8
public WorkflowDefinition Method { get; }
9
private readonly HashSet<string> selection;
10
private readonly Dictionary<(string Node, string Port), string> inputs = [], outputs = [];
11
private readonly string entry;
12
private readonly (string Node, string Port)? exit;
13
public NodeEncapsulation(GraphDefinition graph, IEnumerable<string> ids, IReadOnlyDictionary<string, CapabilityDescriptor> catalog, string name)
14
{
15
selection = ids.ToHashSet(); var chosen = graph.Nodes.Where(x => selection.Contains(x.Id)).ToArray();
16
if (chosen.Length == 0 || chosen.Any(x => x.CapabilityId is "core.start" or "core.end" || WorkflowTriggerNodes.IsTrigger(x)))
17
throw new ArgumentException("请选择动作节点进行封装，开始、结束和触发节点保留在原工作流中");
18
var variableWrites = chosen.Where(x => x.CapabilityId == "core.set").Select(x => x.Parameters.String("name")).ToHashSet();
19
bool UsesVariable(JsonNode? value, Func<string, bool> predicate) => value is JsonObject obj
20
? (obj.Count == 1 && obj["$var"] is { } v && predicate(v.Text().Split('.')[0])) || obj.Any(x => UsesVariable(x.Value, predicate))
21
: value is JsonArray array && array.Any(x => UsesVariable(x, predicate));
22
if (chosen.Any(x => UsesVariable(x.Parameters, name => !variableWrites.Contains(name))) || graph.Nodes.Where(x => !selection.Contains(x.Id)).Any(x => UsesVariable(x.Parameters, variableWrites.Contains)))
23
throw new ArgumentException("选区与外部共享了变量，请先改成数据端口或参数 / 返回值再封装，以保留变量作用域");
24
var incoming = graph.Edges.Where(x => x.Kind == PortKind.Execution && !selection.Contains(x.FromNode) && selection.Contains(x.ToNode)).Select(x => x.ToNode).Distinct().ToArray();
25
var roots = chosen.Where(x => !graph.Edges.Any(e => e.Kind == PortKind.Execution && e.ToNode == x.Id && selection.Contains(e.FromNode))).Select(x => x.Id).ToArray();
26
if (incoming.Length > 1 || roots.Length != 1 || incoming.Length == 1 && incoming[0] != roots[0]) throw new ArgumentException("封装节点需要一个执行入口，请选择一段完整的流程");
27
entry = roots[0];
28
var outgoing = graph.Edges.Where(x => x.Kind == PortKind.Execution && selection.Contains(x.FromNode) && !selection.Contains(x.ToNode)).Select(x => (Node: x.FromNode, Port: x.FromPort)).Distinct().ToArray();
29
if (outgoing.Length > 1) throw new ArgumentException("封装节点需要一个执行出口，请将分支汇合后一起选择");
30
exit = outgoing.Length == 1 ? outgoing[0] : null;
31
if (exit is null)
32
{
33
var tails = chosen.Where(x => !graph.Edges.Any(e => e.Kind == PortKind.Execution && e.FromNode == x.Id && selection.Contains(e.ToNode))).ToArray();
34
if (tails.Length != 1 || catalog[tails[0].CapabilityId].ExecutionOutputs.Count != 1) throw new ArgumentException("请先将执行分支汇合，再封装节点");
35
exit = (tails[0].Id, catalog[tails[0].CapabilityId].ExecutionOutputs[0]);
36
}
37
Method = new() { Name = name, Graph = WorkflowJson.Clone(new GraphDefinition { Nodes = chosen.ToList(), Edges = graph.Edges.Where(x => selection.Contains(x.FromNode) && selection.Contains(x.ToNode)).ToList() }) };
38
string Add((string Node, string Port) source, bool output)
39
{
40
var map = output ? outputs : inputs; if (map.TryGetValue(source, out var known)) return known;
41
var key = (output ? "out" : "arg") + (map.Count + 1); map[source] = key;
42
var node = graph.Nodes.Single(x => x.Id == source.Node);
43
var port = catalog[node.CapabilityId].Outputs.FirstOrDefault(x => x.Name == source.Port);
44
(output ? Method.Outputs : Method.Inputs).Add(new() { Name = key, Label = (node.Name.Length > 0 ? node.Name : catalog[node.CapabilityId].Name) + " · " + (port?.Label ?? source.Port), Type = port?.Type ?? WorkflowValueType.Any }); return key;
45
}
46
foreach (var edge in graph.Edges.Where(x => x.Kind == PortKind.Data && !selection.Contains(x.FromNode) && selection.Contains(x.ToNode)))
47
Method.Graph.Nodes.Single(x => x.Id == edge.ToNode).Parameters[edge.ToPort] = new JsonObject { ["$input"] = Add((edge.FromNode, edge.FromPort), false) };
48
foreach (var edge in graph.Edges.Where(x => x.Kind == PortKind.Data && selection.Contains(x.FromNode) && !selection.Contains(x.ToNode))) Add((edge.FromNode, edge.FromPort), true);
49
foreach (var node in Method.Graph.Nodes) Rewrite(node.Parameters, (source, port, suffix) => !selection.Contains(source) ? new JsonObject { ["$input"] = Add((source, port), false) + suffix } : null);
50
foreach (var node in graph.Nodes.Where(x => !selection.Contains(x.Id))) Scan(node.Parameters, (source, port, _) => { if (selection.Contains(source)) Add((source, port), true); });
51
var start = new NodeDefinition { CapabilityId = "core.start", Name = "参数", X = chosen.Min(x => x.X) - 320, Y = chosen.Min(x => x.Y) };
52
var end = new NodeDefinition { CapabilityId = "core.end", Name = "返回值", X = chosen.Max(x => x.X) + 320, Y = chosen.Max(x => x.Y) };
53
foreach (var pair in outputs) end.Parameters[pair.Value] = new JsonObject { ["$node"] = pair.Key.Node + "." + pair.Key.Port };
54
if (outputs.Count == 0) end.Parameters["result"] = new JsonObject();
55
Method.Graph.Nodes.Add(start); Method.Graph.Nodes.Add(end); Method.Graph.Edges.Add(new() { FromNode = start.Id, ToNode = entry });
56
Method.Graph.Edges.Add(new() { FromNode = exit.Value.Node, FromPort = exit.Value.Port, ToNode = end.Id });
57
}
58
public GraphDefinition Replace(GraphDefinition original, string capability)
59
{
60
var result = WorkflowJson.Clone(original); var chosen = result.Nodes.Where(x => selection.Contains(x.Id)).ToArray();
61
var replacement = new NodeDefinition { CapabilityId = capability, Name = Method.Name, X = chosen.Min(x => x.X), Y = chosen.Min(x => x.Y) };
62
foreach (var pair in inputs) replacement.Parameters[pair.Value] = new JsonObject { ["$node"] = pair.Key.Node + "." + pair.Key.Port };
63
result.Edges.RemoveAll(x => selection.Contains(x.FromNode) && selection.Contains(x.ToNode));
64
foreach (var edge in result.Edges)
65
{
66
if (selection.Contains(edge.ToNode)) { edge.ToNode = replacement.Id; if (edge.Kind == PortKind.Data) edge.ToPort = inputs[(edge.FromNode, edge.FromPort)]; }
67
if (selection.Contains(edge.FromNode)) { var node = edge.FromNode; edge.FromNode = replacement.Id; edge.FromPort = edge.Kind == PortKind.Execution ? "next" : outputs[(node, edge.FromPort)]; }
68
}
69
result.Nodes.RemoveAll(x => selection.Contains(x.Id));
70
foreach (var node in result.Nodes) Rewrite(node.Parameters, (source, port, suffix) => selection.Contains(source) ? new JsonObject { ["$node"] = replacement.Id + "." + outputs[(source, port)] + suffix } : null);
71
result.Nodes.Add(replacement); return result;
72
}
73
private static void Scan(JsonNode? value, Action<string, string, string> action) => Rewrite(value, (node, port, suffix) => { action(node, port, suffix); return null; });
74
private static void Rewrite(JsonNode? value, Func<string, string, string, JsonObject?> replace)
75
{
76
if (value is JsonObject obj)
77
{
78
if (obj.Count == 1 && obj["$node"] is JsonValue text)
79
{
80
var parts = text.GetValue<string>().Split('.', 3);
81
if (parts.Length >= 2 && replace(parts[0], parts[1], parts.Length == 3 ? "." + parts[2] : "") is { } next) { obj.Clear(); foreach (var pair in next) obj[pair.Key] = pair.Value?.DeepClone(); }
82
}
83
else foreach (var child in obj.ToArray()) Rewrite(child.Value, replace);
84
}
85
else if (value is JsonArray array) foreach (var child in array) Rewrite(child, replace);
86
}
87
}
@@ -13,6 +13,10 @@ public sealed class WorkflowEngine(ICodeWorkflowExecutor code, IReadOnlyDictiona
13
13
{
14
14
public async ValueTask<JsonObject> ExecuteAsync(WorkflowDefinition workflow, WorkflowExecutionContext context, CancellationToken cancellationToken)
15
15
{
16
if (context.CallDepth >= 32) throw new InvalidOperationException("封装节点调用不能超过 32 层");
17
var workflowCatalog = WorkflowPortCatalog.For(workflow, catalog);
18
context.WorkflowExecutor = (method, inputs, execution, cancellation) =>
19
new WorkflowEngine(code, catalog).ExecuteAsync(method, execution.MethodScope(method, inputs), cancellation);
16
20
WorkflowTriggerNodes.Migrate(workflow);
17
21
var errors = GraphValidator.Validate(workflow, catalog);
18
22
if (errors.Count > 0) throw new WorkflowExecutionException(string.Join(Environment.NewLine, errors.Select(x => x.Message)), errors[0].NodeId);
@@ -31,11 +35,17 @@ public sealed class WorkflowEngine(ICodeWorkflowExecutor code, IReadOnlyDictiona
31
35
var graphErrors = GraphValidator.Validate(new WorkflowDefinition { Id = workflow.Id, Name = workflow.Name, Graph = graph }, catalog);
32
36
if (graphErrors.Count > 0) throw new WorkflowExecutionException(string.Join(Environment.NewLine, graphErrors.Select(x => x.Message)), graphErrors[0].NodeId);
33
37
var graphEntry = execution.NodeId is null && graph.Nodes.Any(x => x.Id == execution.EntryNodeId) ? execution.EntryNodeId : null;
34
return await ExecuteGraphAsync(graph, execution, cancellation, graphEntry);
38
return await new WorkflowEngine(code, workflowCatalog).ExecuteGraphAsync(graph, execution, cancellation, graphEntry);
35
39
};
36
return workflow.Kind == WorkflowKind.CSharp
40
var outputs = workflow.Kind == WorkflowKind.CSharp
37
41
? await code.ExecuteAsync(workflow.Code, false, workflow.Name + ".cs", context, cancellationToken)
38
: await ExecuteGraphAsync(workflow.Graph, context, cancellationToken, context.EntryNodeId);
42
: await new WorkflowEngine(code, workflowCatalog).ExecuteGraphAsync(workflow.Graph, context, cancellationToken, context.EntryNodeId);
43
foreach (var port in workflow.Outputs)
44
{
45
if (!outputs.ContainsKey(port.Name)) throw new WorkflowExecutionException("缺少返回值：" + port.Name, null);
46
if (outputs[port.Name] is { } value && !GraphValidator.Matches(value, port.Type)) throw new WorkflowExecutionException("返回值类型错误：" + port.Name, null);
47
}
48
return outputs;
39
49
}
40
50
41
51
private async ValueTask<JsonObject> ExecuteGraphAsync(GraphDefinition graph, WorkflowExecutionContext context, CancellationToken ct, string? entryNodeId = null)
@@ -45,7 +55,7 @@ public sealed class WorkflowEngine(ICodeWorkflowExecutor code, IReadOnlyDictiona
45
55
: graph.Nodes.SingleOrDefault(x => x.Id == entryNodeId && WorkflowTriggerNodes.IsTrigger(x));
46
56
if (node is null) throw new WorkflowExecutionException(entryNodeId is null ? "此工作流没有手动入口，请等待触发事件或添加开始节点" : "触发入口不存在", entryNodeId);
47
57
// Legacy graphs can share bindings from the manual entry across triggers.
48
foreach (var start in graph.Nodes.Where(x => x.CapabilityId == "core.start")) context.NodeOutputs[start.Id] = new() { ["inputs"] = context.Inputs.DeepClone() };
58
foreach (var start in graph.Nodes.Where(x => x.CapabilityId == "core.start")) context.NodeOutputs[start.Id] = StartOutputs(context);
49
59
JsonObject result = [];
50
60
while (true)
51
61
{
@@ -106,7 +116,7 @@ public sealed class WorkflowEngine(ICodeWorkflowExecutor code, IReadOnlyDictiona
106
116
node.Children.TryGetValue(name, out var graph) ? await ExecuteGraphAsync(graph, childContext ?? context, ct) : new JsonObject();
107
117
switch (node.CapabilityId)
108
118
{
109
case "core.start": return (new() { ["inputs"] = context.Inputs.DeepClone() }, "next");
119
case "core.start": return (StartOutputs(context), "next");
110
120
case "core.end": return (input, "");
111
121
case "core.log":
112
122
await context.LogAsync(input["message"].Text(), ct); return (input, "next");
@@ -183,9 +193,15 @@ public sealed class WorkflowEngine(ICodeWorkflowExecutor code, IReadOnlyDictiona
183
193
finally { if (!ct.IsCancellationRequested) await Child("finally"); }
184
194
return (new() { ["result"] = result }, "next");
185
195
case "core.join": return (input, "next");
186
default: return (await context.InvokeAsync(node.CapabilityId, input, ct), "next");
196
default:
197
var response = await context.InvokeAsync(node.CapabilityId, input, ct);
198
var port = catalog[node.CapabilityId].ExecutionResultPort is { } key ? response.String(key) : "next";
199
if (!catalog[node.CapabilityId].ExecutionOutputs.Contains(port)) throw new InvalidOperationException("插件返回了未声明的执行分支：" + port);
200
return (response, port);
187
201
}
188
202
}
203
private static JsonObject StartOutputs(WorkflowExecutionContext context)
204
{ var result = WorkflowJson.Copy(context.Inputs); result["inputs"] = context.Inputs.DeepClone(); return result; }
189
205
190
206
public static async ValueTask<JsonNode?> ResolveAsync(JsonNode? value, WorkflowExecutionContext context, CancellationToken ct)
191
207
{
@@ -17,18 +17,28 @@ public sealed class WorkflowExecutionContext : IWorkflowContext
17
17
public string? NodeId { get; init; }
18
18
public string? EntryNodeId { get; init; }
19
19
internal Func<GraphDefinition, WorkflowExecutionContext, CancellationToken, ValueTask<JsonObject>>? GraphExecutor { get; set; }
20
internal Func<WorkflowDefinition, JsonObject, WorkflowExecutionContext, CancellationToken, ValueTask<JsonObject>>? WorkflowExecutor { get; set; }
21
internal int CallDepth { get; init; }
20
22
21
23
public WorkflowExecutionContext Fork(JsonObject? inputs = null, bool isolate = false, string? nodeId = null) => new()
22
24
{
23
25
RunId = RunId, Inputs = inputs ?? Inputs, Variables = isolate ? WorkflowJson.Copy(Variables) : Variables,
24
26
Capabilities = Capabilities, Subworkflow = Subworkflow, Secret = Secret, Host = Host, Report = Report,
25
27
NodeOutputs = isolate ? NodeOutputs.ToDictionary(x => x.Key, x => WorkflowJson.Copy(x.Value)) : NodeOutputs,
26
NodeId = nodeId ?? NodeId, EntryNodeId = EntryNodeId, GraphExecutor = GraphExecutor
28
NodeId = nodeId ?? NodeId, EntryNodeId = EntryNodeId, GraphExecutor = GraphExecutor, WorkflowExecutor = WorkflowExecutor, CallDepth = CallDepth
29
};
30
internal WorkflowExecutionContext MethodScope(WorkflowDefinition workflow, JsonObject inputs) => new()
31
{
32
RunId = RunId, Inputs = WorkflowJson.Copy(inputs), Variables = WorkflowJson.Copy(workflow.Variables),
33
Capabilities = Capabilities, Subworkflow = Subworkflow, Secret = Secret, Host = Host, Report = Report,
34
CallDepth = CallDepth + 1
27
35
};
28
36
public ValueTask<JsonObject> InvokeAsync(string capabilityId, JsonObject inputs, CancellationToken cancellationToken = default) => Capabilities.InvokeAsync(capabilityId, inputs, this, cancellationToken);
29
37
public ValueTask<JsonObject> InvokeWorkflowAsync(string workflowId, JsonObject inputs, CancellationToken cancellationToken = default) => Subworkflow(workflowId, inputs, cancellationToken);
30
38
public ValueTask<JsonObject> ExecuteGraphAsync(GraphDefinition graph, CancellationToken cancellationToken = default)
31
39
=> GraphExecutor?.Invoke(graph, this, cancellationToken) ?? throw new InvalidOperationException("当前上下文尚未连接工作流引擎");
40
public ValueTask<JsonObject> ExecuteWorkflowAsync(WorkflowDefinition workflow, JsonObject inputs, CancellationToken cancellationToken = default)
41
=> WorkflowExecutor?.Invoke(workflow, inputs, this, cancellationToken) ?? throw new InvalidOperationException("当前上下文尚未连接工作流引擎");
32
42
public ValueTask<string> GetSecretAsync(string reference, CancellationToken cancellationToken = default) => Secret(reference, cancellationToken);
33
43
public ValueTask<JsonObject> InvokeHostAsync(string method, JsonObject parameters, CancellationToken cancellationToken = default) => Host(method, parameters, cancellationToken);
34
44
public ValueTask LogAsync(string message, CancellationToken cancellationToken = default)
@@ -0,0 +1,17 @@
1
using XFEWorkflow.Abstractions;
2
3
namespace XFEWorkflow.Core.Runtime;
4
5
public static class WorkflowPortCatalog
6
{
7
public static IReadOnlyDictionary<string, CapabilityDescriptor> For(WorkflowDefinition workflow, IReadOnlyDictionary<string, CapabilityDescriptor> catalog)
8
{
9
var result = catalog.ToDictionary(x => x.Key, x => x.Value);
10
var start = WorkflowJson.Clone(CoreCapabilities.All.Single(x => x.Id == "core.start")); start.Outputs.AddRange(WorkflowJson.Clone(workflow.Inputs)); result[start.Id] = start;
11
if (workflow.Outputs.Count > 0)
12
{
13
var end = WorkflowJson.Clone(CoreCapabilities.All.Single(x => x.Id == "core.end")); end.Inputs = WorkflowJson.Clone(workflow.Outputs); result[end.Id] = end;
14
}
15
return result;
16
}
17
}
@@ -24,6 +24,7 @@ public sealed class CodeCompiler : ICodeWorkflowExecutor, IDisposable
24
24
private readonly List<AssemblyLoadContext> contexts = [];
25
25
private readonly Lazy<List<MetadataReference>> references;
26
26
private static readonly CSharpParseOptions ParseOptions = new(LanguageVersion.CSharp14);
27
public string CapabilityBindingsSource { get; set; } = "";
27
28
public CodeCompiler(string? referenceDirectory = null) => references = new(() => CreateReferences(referenceDirectory));
28
29
public static string WrapNode(string source, string fileName = "node.cs") =>
29
30
"using System; using System.Linq; using System.Threading; using System.Threading.Tasks; using System.Text.Json.Nodes; using XFEWorkflow.Abstractions;\n" +
@@ -43,7 +44,8 @@ public sealed class CodeCompiler : ICodeWorkflowExecutor, IDisposable
43
44
44
45
private CSharpCompilation CreateCompilation(string source, bool body, string file) =>
45
46
CSharpCompilation.Create("XFEWorkflowCode_" + Guid.NewGuid().ToString("N"),
46
[CSharpSyntaxTree.ParseText(SourceText.From(body ? WrapNode(source, file) : source, Encoding.UTF8), ParseOptions, file)],
47
new[] { CSharpSyntaxTree.ParseText(SourceText.From(body ? WrapNode(source, file) : source, Encoding.UTF8), ParseOptions, file) }
48
.Concat(CapabilityBindingsSource.Length == 0 ? [] : new[] { CSharpSyntaxTree.ParseText(SourceText.From(CapabilityBindingsSource, Encoding.UTF8), ParseOptions, "PluginMethods.g.cs") }),
47
49
references.Value, new CSharpCompilationOptions(OutputKind.DynamicallyLinkedLibrary, optimizationLevel: OptimizationLevel.Release, nullableContextOptions: NullableContextOptions.Enable));
48
50
public Task<List<WorkflowDiagnostic>> AnalyzeAsync(string source, bool body, string file, CancellationToken ct = default) => Task.Run(() =>
49
51
CreateCompilation(source, body, file).GetDiagnostics(ct).Where(x => x.Severity is DiagnosticSeverity.Error or DiagnosticSeverity.Warning).Select(ConvertDiagnostic).ToList(), ct);
@@ -55,7 +57,7 @@ public sealed class CodeCompiler : ICodeWorkflowExecutor, IDisposable
55
57
public async ValueTask<JsonObject> ExecuteAsync(string source, bool isNodeBody, string fileName, IWorkflowContext context, CancellationToken ct)
56
58
{
57
59
ct.ThrowIfCancellationRequested();
58
var key = (isNodeBody ? "body:" : "workflow:") + fileName + ":" + source;
60
var key = (isNodeBody ? "body:" : "workflow:") + fileName + ":" + source + CapabilityBindingsSource;
59
61
if (!compiled.TryGetValue(key, out var type))
60
62
{
61
63
var compilation = CreateCompilation(source, isNodeBody, fileName);
@@ -89,12 +91,21 @@ public sealed class CodeCompiler : ICodeWorkflowExecutor, IDisposable
89
91
if (body) position += WrapNode("").IndexOf('\n', prefixLength) + 1;
90
92
var project = workspace.AddProject(ProjectInfo.Create(ProjectId.CreateNewId(), VersionStamp.Create(), "Workflow", "Workflow", LanguageNames.CSharp,
91
93
parseOptions: ParseOptions, compilationOptions: new CSharpCompilationOptions(OutputKind.DynamicallyLinkedLibrary), metadataReferences: references.Value));
94
if (CapabilityBindingsSource.Length > 0) workspace.AddDocument(project.Id, "PluginMethods.g.cs", SourceText.From(CapabilityBindingsSource));
92
95
var document = workspace.AddDocument(project.Id, "workflow.cs", SourceText.From(text));
93
96
var service = CompletionService.GetService(document);
94
97
if (service is null) return [];
95
98
var list = await service.GetCompletionsAsync(document, Math.Clamp(position, 0, text.Length), cancellationToken: ct);
96
99
return list?.ItemsList.Take(150).Select(x => new CodeCompletion(x.DisplayText, x.Properties.GetValueOrDefault("InsertionText", x.DisplayText))).ToList() ?? [];
97
100
}
101
public async Task CompileLibraryAsync(string source, string destination, CancellationToken ct = default)
102
{
103
var compilation = CreateCompilation(source, false, "plugin.g.cs");
104
using var dll = new MemoryStream();
105
var result = compilation.Emit(dll, cancellationToken: ct);
106
if (!result.Success) throw new WorkflowCompilationException(result.Diagnostics.Where(x => x.Severity == DiagnosticSeverity.Error).Select(ConvertDiagnostic).ToList());
107
await File.WriteAllBytesAsync(destination, dll.ToArray(), ct);
108
}
98
109
public void Dispose()
99
110
{
100
111
compiled.Clear();
@@ -0,0 +1,42 @@
1
using System.Text;
2
using System.Text.Json;
3
using System.Text.RegularExpressions;
4
using XFEWorkflow.Abstractions;
5
6
namespace XFEWorkflow.Infrastructure.Compilation;
7
8
/// <summary>Produces SDK-only method adapters; implementation assemblies are never referenced.</summary>
9
public static class PluginMethodBindings
10
{
11
public static string Identifier(string name) => "_" + Regex.Replace(name, "[^\\p{L}\\p{Nd}_]", "_");
12
public static string MethodName(CapabilityDescriptor capability) => Identifier(capability.Id[(capability.PluginId.Length + 1)..]) + "Async";
13
public static string CallName(CapabilityDescriptor capability) => "XfePlugins." + Identifier(capability.PluginId) + "." + MethodName(capability);
14
private static string Type(PortDefinition port) => port.Type switch
15
{
16
WorkflowValueType.String or WorkflowValueType.Resource => "string?", WorkflowValueType.Number => "double?", WorkflowValueType.Boolean => "bool?",
17
WorkflowValueType.Object => "JsonObject?", WorkflowValueType.Array => "JsonArray?", _ => "JsonNode?"
18
};
19
public static string Generate(IEnumerable<CapabilityDescriptor> capabilities)
20
{
21
var source = new StringBuilder("#nullable enable\nusing System; using System.Threading; using System.Threading.Tasks; using System.Text.Json; using System.Text.Json.Nodes; using XFEWorkflow.Abstractions;\nnamespace XfePlugins {\n");
22
foreach (var group in capabilities.Where(x => !x.IsTrigger && x.PluginId != "core").GroupBy(x => x.PluginId).OrderBy(x => x.Key))
23
{
24
source.Append("public static class ").Append(Identifier(group.Key)).Append(" {\n");
25
foreach (var capability in group.OrderBy(x => x.Id))
26
{
27
var outputs = capability.Outputs;
28
var resultType = outputs.Count switch { 0 => "", 1 => "<" + Type(outputs[0]) + ">", _ => "<(" + string.Join(", ", outputs.Select(x => Type(x) + " " + Identifier(x.Name))) + ")>" };
29
source.Append("public static async ValueTask").Append(resultType).Append(' ').Append(MethodName(capability)).Append("(IWorkflowContext context");
30
foreach (var input in capability.Inputs) source.Append(", ").Append(Type(input)).Append(' ').Append(Identifier(input.Name));
31
source.Append(", CancellationToken cancellationToken = default) {\nvar args = new JsonObject();\n");
32
foreach (var input in capability.Inputs) source.Append("args[").Append(JsonSerializer.Serialize(input.Name)).Append("] = JsonSerializer.SerializeToNode(").Append(Identifier(input.Name)).Append(");\n");
33
source.Append("var result = await context.InvokeAsync(").Append(JsonSerializer.Serialize(capability.Id)).Append(", args, cancellationToken);\n");
34
var expressions = outputs.Select(x => "result[" + JsonSerializer.Serialize(x.Name) + "]?.Deserialize<" + Type(x) + ">()" ).ToList();
35
if (outputs.Count > 0) source.Append("return ").Append(outputs.Count == 1 ? expressions[0] : "(" + string.Join(", ", expressions) + ")").Append(';');
36
source.Append("\n}\n");
37
}
38
source.Append("}\n");
39
}
40
return source.Append("}\n").ToString();
41
}
42
}
@@ -0,0 +1,89 @@
1
using System.IO.Compression;
2
using System.Text.Json;
3
using XFEWorkflow.Abstractions;
4
using XFEWorkflow.Core.Runtime;
5
using XFEWorkflow.Infrastructure.Compilation;
6
using XFEWorkflow.Infrastructure.Storage;
7
8
namespace XFEWorkflow.Infrastructure.Plugins;
9
10
public sealed class GraphPluginProject
11
{
12
public string Id { get; set; } = "user.myplugin";
13
public string Name { get; set; } = "我的插件";
14
public string Version { get; set; } = "1.0.0";
15
public string Author { get; set; } = "";
16
public string Description { get; set; } = "";
17
public List<GraphPluginMethod> Methods { get; set; } = [];
18
}
19
public sealed class GraphPluginMethod
20
{
21
public string Id { get; set; } = "run";
22
public string Name { get; set; } = "自定义节点";
23
public string Category { get; set; } = "自定义节点";
24
public WorkflowDefinition Workflow { get; set; } = new();
25
}
26
27
public sealed class GraphPluginBuilder(PluginManager plugins)
28
{
29
public PluginManifest Describe(GraphPluginProject project)
30
{
31
if (project.Methods.Count is < 1 or > 128 || project.Methods.Select(x => x.Id).Distinct().Count() != project.Methods.Count)
32
throw new ArgumentException("一个插件需要 1 到 128 个方法，方法 ID 不能重复");
33
var manifest = new PluginManifest { Id = project.Id, Name = project.Name, Version = project.Version, Author = project.Author, Description = project.Description,
34
SdkMinimumVersion = "1.3.0", Assembly = "XFEWorkflow.GeneratedPlugin.dll", EntryType = "XFEWorkflow.Generated.Plugin" };
35
var required = new Dictionary<string, InstalledPlugin>();
36
foreach (var method in project.Methods)
37
{
38
WorkspacePaths.RequireId(method.Id);
39
var workflow = method.Workflow;
40
if (WorkflowTriggerNodes.Enumerate(workflow).Any() || workflow.Triggers.Count > 0) throw new ArgumentException("方法不能包含自动触发节点；请保留触发器，在其后封装动作");
41
var errors = GraphValidator.Validate(workflow, plugins.Catalog(workflow));
42
if (errors.Any(x => x.IsError)) throw new ArgumentException(string.Join("\n", errors.Select(x => x.Message)));
43
foreach (var dependency in plugins.Resolve(workflow))
44
{
45
if (dependency.Manifest.Id == project.Id) throw new ArgumentException("插件方法不能依赖自身");
46
if (required.TryGetValue(dependency.Manifest.Id, out var previous) && previous.Manifest.Version != dependency.Manifest.Version)
47
throw new ArgumentException("不同方法的依赖版本冲突：" + dependency.Manifest.Id);
48
required[dependency.Manifest.Id] = dependency;
49
}
50
manifest.Capabilities.Add(new() { Id = project.Id + "." + method.Id, PluginId = project.Id, Name = method.Name, Category = method.Category,
51
Description = workflow.Description, Inputs = WorkflowJson.Clone(workflow.Inputs), Outputs = WorkflowJson.Clone(workflow.Outputs.Count > 0 ? workflow.Outputs : [new PortDefinition { Name = "result", Label = "返回值", Type = WorkflowValueType.Any }]) });
52
}
53
manifest.Dependencies = required.Values.OrderBy(x => x.Manifest.Id).Select(x => new PluginDependency { Id = x.Manifest.Id, MinimumVersion = x.Manifest.Version, ExactVersion = x.Manifest.Version }).ToList();
54
manifest.Permissions = required.Values.SelectMany(x => x.Manifest.Permissions).Distinct().Order().ToList();
55
PluginPackage.Validate(manifest); return manifest;
56
}
57
public async Task<PluginManifest> BuildAsync(GraphPluginProject project, string destination, CancellationToken ct = default)
58
{
59
var manifest = Describe(project); var folder = Path.Combine(Path.GetTempPath(), "XFEWorkflow-PluginBuild", Guid.NewGuid().ToString("N")); Directory.CreateDirectory(folder);
60
try
61
{
62
var registrations = string.Join("\n", project.Methods.Select((method, i) => "registry.RegisterAction(manifest.Capabilities[" + i + "], () => new Method(" + Literal(JsonSerializer.Serialize(method.Workflow, WorkflowJson.Options)) + "));"));
63
var source = """
64
using System.Threading; using System.Threading.Tasks; using System.Text.Json; using System.Text.Json.Nodes; using XFEWorkflow.Abstractions;
65
namespace XFEWorkflow.Generated {
66
public sealed class Plugin : IWorkflowPlugin {
67
public void Register(IPluginRegistry registry) {
68
""" + "\nvar manifest = JsonSerializer.Deserialize<PluginManifest>(" + Literal(JsonSerializer.Serialize(manifest, WorkflowJson.Options)) + ", WorkflowJson.Options)!;\n" + registrations + """
69
}
70
}
71
public sealed class Method(string definition) : IWorkflowAction {
72
public ValueTask<JsonObject> ExecuteAsync(JsonObject inputs, IWorkflowContext context, CancellationToken cancellationToken)
73
=> context.ExecuteWorkflowAsync(JsonSerializer.Deserialize<WorkflowDefinition>(definition, WorkflowJson.Options)!, inputs, cancellationToken);
74
}
75
}
76
""";
77
using var compiler = new CodeCompiler(); await compiler.CompileLibraryAsync(source, Path.Combine(folder, manifest.Assembly), ct);
78
await AtomicJson.WriteAsync(Path.Combine(folder, "plugin.json"), manifest, ct);
79
await AtomicJson.WriteAsync(Path.Combine(folder, "project.json"), project, ct);
80
await File.WriteAllTextAsync(Path.Combine(folder, "PluginMethods.g.cs"), PluginMethodBindings.Generate(manifest.Capabilities), ct);
81
var staged = Path.Combine(folder, "package.xfeplugin");
82
using (var archive = ZipFile.Open(staged, ZipArchiveMode.Create))
83
foreach (var file in Directory.EnumerateFiles(folder).Where(x => x != staged)) archive.CreateEntryFromFile(file, Path.GetFileName(file));
84
PluginPackage.Inspect(staged); File.Move(staged, destination, false); return manifest;
85
}
86
finally { Directory.Delete(folder, true); }
87
}
88
private static string Literal(string value) => JsonSerializer.Serialize(value);
89
}
@@ -0,0 +1,36 @@
1
using XFEWorkflow.Abstractions;
2
3
namespace XFEWorkflow.Infrastructure.Plugins;
4
5
public static class PluginDependencyResolver
6
{
7
public static async Task<List<PluginManifest>> ResolveAsync(PluginManifest root, Func<string, CancellationToken, Task<List<PluginManifest>>> versions, CancellationToken ct = default)
8
{
9
var cache = new Dictionary<string, List<PluginManifest>>(); var attempts = 0; var visited = new HashSet<string>();
10
async Task<Dictionary<string, PluginManifest>?> Search(Dictionary<string, PluginManifest> chosen)
11
{
12
ct.ThrowIfCancellationRequested(); if (++attempts > 10000 || chosen.Count > 128) throw new InvalidOperationException("插件依赖过于复杂");
13
var requirements = chosen.Values.SelectMany(x => x.Dependencies).GroupBy(x => x.Id).ToDictionary(x => x.Key, x => x.ToList());
14
foreach (var pair in requirements) { visited.Add(pair.Key); if (chosen.TryGetValue(pair.Key, out var selected) && !pair.Value.All(x => x.Accepts(selected.Version))) return null; }
15
var pending = requirements.Keys.Where(x => !chosen.ContainsKey(x)).Order(StringComparer.Ordinal).FirstOrDefault();
16
if (pending is null) return HasCycle(chosen) ? null : chosen;
17
if (!cache.TryGetValue(pending, out var candidates)) cache[pending] = candidates = await versions(pending, ct);
18
foreach (var candidate in candidates.Where(x => x.Id == pending && requirements[pending].All(r => r.Accepts(x.Version))).OrderByDescending(x => SemanticVersion.Parse(x.Version)))
19
{
20
PluginPackage.Validate(candidate); var next = new Dictionary<string, PluginManifest>(chosen) { [pending] = candidate };
21
if (await Search(next) is { } solution) return solution;
22
}
23
return null;
24
}
25
var result = await Search(new() { [root.Id] = root }) ?? throw new InvalidOperationException("依赖缺失、版本冲突或循环引用：" + string.Join("、", visited.Order()));
26
List<PluginManifest> ordered = []; HashSet<string> added = [];
27
void Add(PluginManifest manifest) { if (!added.Add(manifest.Id)) return; foreach (var dependency in manifest.Dependencies) Add(result[dependency.Id]); ordered.Add(manifest); }
28
Add(root); return ordered;
29
}
30
private static bool HasCycle(Dictionary<string, PluginManifest> chosen)
31
{
32
HashSet<string> active = [], complete = [];
33
bool Visit(string id) { if (active.Contains(id)) return true; if (!complete.Add(id)) return false; active.Add(id); foreach (var dependency in chosen[id].Dependencies) if (Visit(dependency.Id)) return true; active.Remove(id); return false; }
34
return chosen.Keys.Any(Visit);
35
}
36
}
@@ -21,7 +21,8 @@ public sealed class PluginManager(WorkspacePaths paths, ClientState state)
21
21
workflow?.PluginVersions.TryGetValue(g.Key, out var version) == true
22
22
? g.FirstOrDefault(x => x.Manifest.Version == version)
23
23
: g.MaxBy(x => SemanticVersion.Parse(x.Manifest.Version))).OfType<InstalledPlugin>();
24
return CoreCapabilities.Catalog(selected.SelectMany(x => x.Manifest.Capabilities));
24
var catalog = CoreCapabilities.Catalog(selected.SelectMany(x => x.Manifest.Capabilities));
25
return workflow is null ? catalog : WorkflowPortCatalog.For(workflow, catalog);
25
26
}
26
27
public async Task ImportBundledAsync(string root, CancellationToken ct = default)
27
28
{
@@ -143,9 +144,7 @@ public sealed class PluginManager(WorkspacePaths paths, ClientState state)
143
144
}
144
145
return resolved.Values.ToList();
145
146
}
146
private static bool InRange(string version, PluginDependency dependency) =>
147
SemanticVersion.Parse(version).CompareTo(SemanticVersion.Parse(dependency.MinimumVersion)) >= 0
148
&& (dependency.MaximumVersionExclusive is null || SemanticVersion.Parse(version).CompareTo(SemanticVersion.Parse(dependency.MaximumVersionExclusive)) < 0);
147
private static bool InRange(string version, PluginDependency dependency) => dependency.Accepts(version);
149
148
private static IEnumerable<string> UsedCapabilities(GraphDefinition graph) =>
150
149
graph.Nodes.Select(x => x.CapabilityId).Concat(graph.Nodes.SelectMany(x => x.Children.Values).SelectMany(UsedCapabilities));
151
150
public async Task<IAsyncDisposable> AcquireAsync(IEnumerable<InstalledPlugin> plugins, CancellationToken ct = default)