457 lines
19 KiB
C#
457 lines
19 KiB
C#
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging;
|
|
using VolcanicCore.Api.Registries;
|
|
using VolcanicCore.Core;
|
|
using VolcanicCore.Persistence.Repositories;
|
|
|
|
namespace VolcanicCore.Api.Engine;
|
|
|
|
public sealed record WorkflowNode(
|
|
string Id, string TypeId, string? ConnectionId, string ConfigJson,
|
|
string? PinnedInputJson = null,
|
|
string? PinnedOutputJson = null);
|
|
public sealed record WorkflowEdge(string FromNode, int FromPort, string ToNode, int ToPort);
|
|
public sealed record WorkflowGraph
|
|
{
|
|
public IReadOnlyList<WorkflowNode> Nodes { get; init; } = Array.Empty<WorkflowNode>();
|
|
public IReadOnlyList<WorkflowEdge> Edges { get; init; } = Array.Empty<WorkflowEdge>();
|
|
|
|
public static WorkflowGraph Parse(string graphJson)
|
|
{
|
|
var opts = new JsonSerializerOptions(JsonSerializerDefaults.Web)
|
|
{
|
|
PropertyNameCaseInsensitive = true,
|
|
};
|
|
using var doc = JsonDocument.Parse(graphJson);
|
|
var root = doc.RootElement;
|
|
var nodes = new List<WorkflowNode>();
|
|
foreach (var n in root.GetProperty("nodes").EnumerateArray())
|
|
{
|
|
var id = n.GetProperty("id").GetString()!;
|
|
var typeId = n.TryGetProperty("type_id", out var t1) ? t1.GetString()
|
|
: n.TryGetProperty("typeId", out var t2) ? t2.GetString() : null;
|
|
var connId = n.TryGetProperty("connection_id", out var c1) ? c1.GetString()
|
|
: n.TryGetProperty("connectionId", out var c2) ? c2.GetString() : null;
|
|
var cfg = n.TryGetProperty("config_json", out var g1) ? g1.GetRawText()
|
|
: n.TryGetProperty("config", out var g2) ? g2.GetRawText() : "{}";
|
|
var pinIn = ReadPin(n, "pinned_input");
|
|
var pinOut = ReadPin(n, "pinned_output");
|
|
nodes.Add(new WorkflowNode(id, typeId!, connId, cfg, pinIn, pinOut));
|
|
}
|
|
var edges = new List<WorkflowEdge>();
|
|
if (root.TryGetProperty("edges", out var edgesEl))
|
|
{
|
|
foreach (var e in edgesEl.EnumerateArray())
|
|
{
|
|
var from = e.GetProperty("from_node").GetString()!;
|
|
var to = e.GetProperty("to_node").GetString()!;
|
|
var fp = e.TryGetProperty("from_port", out var fpe) ? fpe.GetInt32() : 0;
|
|
var tp = e.TryGetProperty("to_port", out var tpe) ? tpe.GetInt32() : 0;
|
|
edges.Add(new WorkflowEdge(from, fp, to, tp));
|
|
}
|
|
}
|
|
return new WorkflowGraph { Nodes = nodes, Edges = edges };
|
|
}
|
|
|
|
private static string? ReadPin(JsonElement node, string name)
|
|
{
|
|
if (!node.TryGetProperty(name, out var v)) return null;
|
|
if (v.ValueKind == JsonValueKind.Null) return null;
|
|
return v.GetRawText();
|
|
}
|
|
}
|
|
|
|
public sealed class Executor
|
|
{
|
|
private readonly ExecutionRepository _exRepo;
|
|
private readonly NodeExecutionRepository _neRepo;
|
|
private readonly NodeRegistry _nodes;
|
|
private readonly IConnectionInstanceRegistry _instances;
|
|
private readonly ILoggerFactory _lf;
|
|
|
|
public Executor(
|
|
ExecutionRepository exRepo,
|
|
NodeExecutionRepository neRepo,
|
|
NodeRegistry nodes,
|
|
IConnectionInstanceRegistry instances,
|
|
ILoggerFactory lf)
|
|
{
|
|
_exRepo = exRepo;
|
|
_neRepo = neRepo;
|
|
_nodes = nodes;
|
|
_instances = instances;
|
|
_lf = lf;
|
|
}
|
|
|
|
public async Task<string> StartExecutionAsync(
|
|
string workflowId,
|
|
WorkflowGraph graph,
|
|
string triggerNodeId,
|
|
string triggerKind,
|
|
IReadOnlyList<Item> triggerItems,
|
|
JsonElement? triggerData,
|
|
Guid? answerChannelId,
|
|
CancellationToken ct)
|
|
{
|
|
var triggerDataJson = triggerData?.GetRawText() ?? "null";
|
|
var eid = _exRepo.Create(workflowId, triggerKind, triggerDataJson);
|
|
var logger = _lf.CreateLogger($"exec.{eid}");
|
|
|
|
try
|
|
{
|
|
await RunAsync(eid, workflowId, graph, triggerNodeId, triggerKind, triggerItems, answerChannelId, logger, ct);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_exRepo.SetStatus(eid, "failed", ex.Message, DateTimeOffset.UtcNow);
|
|
}
|
|
return eid;
|
|
}
|
|
|
|
private enum ArrivalKind { Data, Failure }
|
|
private sealed record Arrival(ArrivalKind Kind, IReadOnlyList<Item> Items);
|
|
|
|
private async Task RunAsync(
|
|
string executionId, string workflowId, WorkflowGraph graph,
|
|
string triggerNodeId, string triggerKind, IReadOnlyList<Item> triggerItems,
|
|
Guid? answerChannelId, ILogger logger, CancellationToken ct)
|
|
{
|
|
var triggerNode = graph.Nodes.FirstOrDefault(n => n.Id == triggerNodeId)
|
|
?? throw new InvalidOperationException($"Trigger node {triggerNodeId} not found in graph.");
|
|
|
|
// Manual runs honor the trigger's pinned_output — replaces the
|
|
// endpoint-passed triggerItems entirely. Live runs (webhook, NATS, etc.)
|
|
// ignore the pin so a stale value never corrupts real traffic.
|
|
if (triggerKind == "manual" && triggerNode.PinnedOutputJson is { } pinJson)
|
|
{
|
|
using var pinDoc = JsonDocument.Parse(pinJson);
|
|
triggerItems = WrapPin(pinDoc.RootElement);
|
|
}
|
|
|
|
var triggerStart = DateTimeOffset.UtcNow;
|
|
_neRepo.Create(executionId, triggerNodeId,
|
|
inputsJson: "[]",
|
|
outputsJson: SerializeOutputs(new[] { triggerItems }),
|
|
status: "completed",
|
|
startedAt: triggerStart,
|
|
endedAt: DateTimeOffset.UtcNow,
|
|
error: null);
|
|
|
|
var arrivals = new Dictionary<string, Dictionary<int, Arrival>>(StringComparer.Ordinal);
|
|
foreach (var n in graph.Nodes) arrivals[n.Id] = new();
|
|
var inDegree = graph.Nodes.ToDictionary(n => n.Id,
|
|
n => graph.Edges.Count(e => e.ToNode == n.Id), StringComparer.Ordinal);
|
|
|
|
var ready = new Queue<string>();
|
|
var running = new Dictionary<Task<NodeResult>, string>();
|
|
var completedCount = 0;
|
|
var failedCount = 0;
|
|
var skippedCount = 0;
|
|
|
|
void Deliver(string toNode, int toPort, Arrival a)
|
|
{
|
|
arrivals[toNode][toPort] = a;
|
|
if (arrivals[toNode].Count == inDegree[toNode]) ready.Enqueue(toNode);
|
|
}
|
|
|
|
foreach (var e in graph.Edges.Where(e => e.FromNode == triggerNodeId && e.FromPort == 0))
|
|
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Data, triggerItems));
|
|
|
|
while (ready.Count > 0 || running.Count > 0)
|
|
{
|
|
while (ready.Count > 0)
|
|
{
|
|
var nid = ready.Dequeue();
|
|
var node = graph.Nodes.First(n => n.Id == nid);
|
|
var portMap = arrivals[nid];
|
|
|
|
// Skip if ANY incoming was a failure.
|
|
if (portMap.Values.Any(v => v.Kind == ArrivalKind.Failure))
|
|
{
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
_neRepo.Create(executionId, nid,
|
|
inputsJson: SerializeInputs(BuildInputsByPort(portMap, allowFailurePlaceholders: true)),
|
|
outputsJson: "[]",
|
|
status: "skipped_due_to_failure",
|
|
startedAt: startedAt,
|
|
endedAt: startedAt,
|
|
error: null);
|
|
skippedCount++;
|
|
foreach (var e in graph.Edges.Where(e => e.FromNode == nid))
|
|
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Failure, Array.Empty<Item>()));
|
|
continue;
|
|
}
|
|
|
|
var byPort = BuildInputsByPort(portMap, allowFailurePlaceholders: false);
|
|
var typ = _nodes.Get(node.TypeId);
|
|
if (typ is not IActionNode action)
|
|
throw new InvalidOperationException($"Node {nid} is not an action type.");
|
|
|
|
running[RunNodeAsync(executionId, workflowId, node, action, byPort, answerChannelId, ct)] = nid;
|
|
}
|
|
|
|
if (running.Count == 0) break;
|
|
var finished = await Task.WhenAny(running.Keys);
|
|
var nidDone = running[finished];
|
|
running.Remove(finished);
|
|
var result = await finished;
|
|
|
|
_neRepo.Create(executionId, nidDone,
|
|
inputsJson: result.InputsJson,
|
|
outputsJson: result.Success ? SerializeOutputs(result.Outputs!) : "[]",
|
|
status: result.Success ? "completed" : "failed",
|
|
startedAt: result.StartedAt,
|
|
endedAt: result.EndedAt,
|
|
error: result.Error);
|
|
|
|
if (result.Success)
|
|
{
|
|
completedCount++;
|
|
foreach (var e in graph.Edges.Where(e => e.FromNode == nidDone))
|
|
{
|
|
if (e.FromPort >= result.Outputs!.Count) continue;
|
|
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Data, result.Outputs[e.FromPort]));
|
|
}
|
|
}
|
|
else
|
|
{
|
|
failedCount++;
|
|
foreach (var e in graph.Edges.Where(e => e.FromNode == nidDone))
|
|
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Failure, Array.Empty<Item>()));
|
|
}
|
|
}
|
|
|
|
// Status roll-up (spec §4):
|
|
// completed = no failed nodes
|
|
// failed = at least one failed + zero non-trigger completions
|
|
// partial = otherwise (failures + successes both present)
|
|
var status = failedCount == 0
|
|
? "completed"
|
|
: (completedCount == 0 ? "failed" : "partial");
|
|
_exRepo.SetStatus(executionId, status, error: null, endedAt: DateTimeOffset.UtcNow);
|
|
}
|
|
|
|
private static IReadOnlyList<IReadOnlyList<Item>> BuildInputsByPort(
|
|
Dictionary<int, Arrival> portMap, bool allowFailurePlaceholders)
|
|
{
|
|
var count = portMap.Count == 0 ? 1 : portMap.Keys.Max() + 1;
|
|
var byPort = new IReadOnlyList<Item>[count];
|
|
for (var p = 0; p < count; p++)
|
|
{
|
|
if (portMap.TryGetValue(p, out var a))
|
|
byPort[p] = a.Items;
|
|
else if (allowFailurePlaceholders)
|
|
byPort[p] = Array.Empty<Item>();
|
|
else
|
|
byPort[p] = Array.Empty<Item>();
|
|
}
|
|
return byPort;
|
|
}
|
|
|
|
private sealed record NodeResult(
|
|
bool Success,
|
|
IReadOnlyList<IReadOnlyList<Item>>? Outputs,
|
|
string InputsJson,
|
|
DateTimeOffset StartedAt,
|
|
DateTimeOffset EndedAt,
|
|
string? Error);
|
|
|
|
private async Task<NodeResult> RunNodeAsync(
|
|
string executionId, string workflowId, WorkflowNode node, IActionNode action,
|
|
IReadOnlyList<IReadOnlyList<Item>> byPort, Guid? answerChannelId, CancellationToken ct)
|
|
{
|
|
var inputsJson = SerializeInputs(byPort);
|
|
using var cfgDoc = JsonDocument.Parse(node.ConfigJson);
|
|
// Templating pass: every string value inside the node's config JSON
|
|
// gets `{input.foo}` interpolation against the first item on port 0
|
|
// BEFORE the node sees it, so every action node gets template support
|
|
// for free. See TemplateEngine + CLAUDE.md "config-field templates".
|
|
var firstInput = byPort.Count > 0 && byPort[0].Count > 0 ? byPort[0][0].Json : (JsonElement?)null;
|
|
var templatedConfig = InterpolateStrings(cfgDoc.RootElement, firstInput);
|
|
// After templating, coerce string leaves that the schema declares as
|
|
// integer/number/boolean back to their typed JSON form. Nodes check
|
|
// ValueKind strictly, so a templated `"{input.code}" -> "201"` string
|
|
// would silently miss the ValueKind.Number check without this pass.
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
JsonElement coercedConfig;
|
|
try
|
|
{
|
|
coercedConfig = CoerceToSchema(templatedConfig, action.ConfigSchema);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
return new NodeResult(false, null, inputsJson, startedAt, DateTimeOffset.UtcNow, ex.Message);
|
|
}
|
|
var ctx = new ExecutionContext(
|
|
executionId, workflowId, node.Id, coercedConfig,
|
|
answerChannelId, _lf.CreateLogger($"node.{node.Id}"),
|
|
node.ConnectionId, _instances);
|
|
|
|
try
|
|
{
|
|
var outputs = await action.ExecuteAsync(ctx, byPort, ct);
|
|
return new NodeResult(true, outputs, inputsJson, startedAt, DateTimeOffset.UtcNow, null);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
return new NodeResult(false, null, inputsJson, startedAt, DateTimeOffset.UtcNow, ex.Message);
|
|
}
|
|
}
|
|
|
|
// Walks the top-level properties of the schema. For each property whose
|
|
// declared type is integer/number/boolean, if the config leaf is a JSON
|
|
// string (post-interpolation), parse it back to its typed form. Missing
|
|
// properties, non-primitive schema types, and string-typed schema fields
|
|
// are left untouched. Throws InvalidOperationException with a readable
|
|
// message on parse failure — the caller surfaces this as a normal node
|
|
// failure with the message stored on the node execution row.
|
|
private static JsonElement CoerceToSchema(JsonElement config, JsonSchema schema)
|
|
{
|
|
if (config.ValueKind != JsonValueKind.Object) return config;
|
|
if (schema.Root.ValueKind != JsonValueKind.Object) return config;
|
|
if (!schema.Root.TryGetProperty("properties", out var props)
|
|
|| props.ValueKind != JsonValueKind.Object) return config;
|
|
|
|
// Fast path: nothing to coerce → skip the rewrite.
|
|
var anyNeedsCoerce = false;
|
|
foreach (var prop in props.EnumerateObject())
|
|
{
|
|
var t = TypeOf(prop.Value);
|
|
if (t is null || t == "string") continue;
|
|
if (!config.TryGetProperty(prop.Name, out var leaf)) continue;
|
|
if (leaf.ValueKind == JsonValueKind.String) { anyNeedsCoerce = true; break; }
|
|
}
|
|
if (!anyNeedsCoerce) return config;
|
|
|
|
var buffer = new System.IO.MemoryStream();
|
|
using (var w = new Utf8JsonWriter(buffer))
|
|
{
|
|
w.WriteStartObject();
|
|
foreach (var p in config.EnumerateObject())
|
|
{
|
|
w.WritePropertyName(p.Name);
|
|
if (props.TryGetProperty(p.Name, out var propSchema)
|
|
&& p.Value.ValueKind == JsonValueKind.String)
|
|
{
|
|
var t = TypeOf(propSchema);
|
|
var s = p.Value.GetString() ?? "";
|
|
switch (t)
|
|
{
|
|
case "integer":
|
|
if (long.TryParse(s, System.Globalization.NumberStyles.Integer,
|
|
System.Globalization.CultureInfo.InvariantCulture, out var i))
|
|
w.WriteNumberValue(i);
|
|
else
|
|
throw new InvalidOperationException(
|
|
$"config '{p.Name}': expected integer, got '{s}'");
|
|
break;
|
|
case "number":
|
|
if (double.TryParse(s, System.Globalization.NumberStyles.Float,
|
|
System.Globalization.CultureInfo.InvariantCulture, out var d))
|
|
w.WriteNumberValue(d);
|
|
else
|
|
throw new InvalidOperationException(
|
|
$"config '{p.Name}': expected number, got '{s}'");
|
|
break;
|
|
case "boolean":
|
|
if (s == "true") w.WriteBooleanValue(true);
|
|
else if (s == "false") w.WriteBooleanValue(false);
|
|
else
|
|
throw new InvalidOperationException(
|
|
$"config '{p.Name}': expected boolean, got '{s}'");
|
|
break;
|
|
default:
|
|
p.Value.WriteTo(w);
|
|
break;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
p.Value.WriteTo(w);
|
|
}
|
|
}
|
|
w.WriteEndObject();
|
|
}
|
|
using var doc = JsonDocument.Parse(buffer.ToArray());
|
|
return doc.RootElement.Clone();
|
|
}
|
|
|
|
private static string? TypeOf(JsonElement schema)
|
|
{
|
|
if (schema.ValueKind != JsonValueKind.Object) return null;
|
|
if (!schema.TryGetProperty("type", out var t) || t.ValueKind != JsonValueKind.String) return null;
|
|
return t.GetString();
|
|
}
|
|
|
|
// Walks the config JSON, running every string leaf through TemplateEngine.
|
|
// Objects/arrays are rebuilt recursively; non-string leaves pass through.
|
|
// Fast-path: if the raw text contains no `{`, no template can fire, so we
|
|
// just clone the element.
|
|
private static JsonElement InterpolateStrings(JsonElement source, JsonElement? input)
|
|
{
|
|
if (source.ValueKind is not (JsonValueKind.Object or JsonValueKind.Array or JsonValueKind.String))
|
|
return source.Clone();
|
|
if (source.GetRawText().IndexOf('{') < 0)
|
|
return source.Clone();
|
|
|
|
var buffer = new System.IO.MemoryStream();
|
|
using (var writer = new Utf8JsonWriter(buffer))
|
|
WriteInterpolated(source, input, writer);
|
|
using var doc = JsonDocument.Parse(buffer.ToArray());
|
|
return doc.RootElement.Clone();
|
|
}
|
|
|
|
private static void WriteInterpolated(JsonElement el, JsonElement? input, Utf8JsonWriter w)
|
|
{
|
|
switch (el.ValueKind)
|
|
{
|
|
case JsonValueKind.Object:
|
|
w.WriteStartObject();
|
|
foreach (var p in el.EnumerateObject())
|
|
{
|
|
w.WritePropertyName(p.Name);
|
|
WriteInterpolated(p.Value, input, w);
|
|
}
|
|
w.WriteEndObject();
|
|
break;
|
|
case JsonValueKind.Array:
|
|
w.WriteStartArray();
|
|
foreach (var item in el.EnumerateArray())
|
|
WriteInterpolated(item, input, w);
|
|
w.WriteEndArray();
|
|
break;
|
|
case JsonValueKind.String:
|
|
w.WriteStringValue(TemplateEngine.Render(el.GetString() ?? "", input));
|
|
break;
|
|
default:
|
|
el.WriteTo(w);
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Wraps a pinned JSON value into a port-shape Item[] on port 0.
|
|
// - Array → one item per element (each item's Json = the element)
|
|
// - Anything else (object, scalar, null) → single item wrapping the value
|
|
// `internal` so unit tests can reach it without going through a full run.
|
|
internal static IReadOnlyList<Item> WrapPin(JsonElement pin)
|
|
{
|
|
if (pin.ValueKind == JsonValueKind.Array)
|
|
{
|
|
var arr = new List<Item>(pin.GetArrayLength());
|
|
foreach (var el in pin.EnumerateArray())
|
|
arr.Add(new Item(el.Clone()));
|
|
return arr;
|
|
}
|
|
return new[] { new Item(pin.Clone()) };
|
|
}
|
|
|
|
private static string SerializeInputs(IReadOnlyList<IReadOnlyList<Item>> ports)
|
|
=> JsonSerializer.Serialize(ports.Select(port => port.Select(i => new
|
|
{
|
|
json = i.Json,
|
|
binary = i.Binary?.ToDictionary(kv => kv.Key, kv => Convert.ToBase64String(kv.Value)),
|
|
}).ToArray()).ToArray());
|
|
|
|
private static string SerializeOutputs(IReadOnlyList<IReadOnlyList<Item>> ports) => SerializeInputs(ports);
|
|
}
|