feat(api): manual run + executions REST (list + detail with per-node ports)

This commit is contained in:
Claude
2026-08-28 15:05:54 +00:00
parent 850eb4a12c
commit 0895dc5f8f
3 changed files with 150 additions and 0 deletions

View File

@@ -0,0 +1,83 @@
using System.Text.Json;
using VolcanicCore.Api.Engine;
using VolcanicCore.Core;
using VolcanicCore.Persistence.Repositories;
namespace VolcanicCore.Api.Endpoints;
public static class ExecutionsEndpoints
{
public sealed record ManualRunRequest(string StartFromNodeId, JsonElement? TriggerData);
public static void MapExecutionsEndpoints(this IEndpointRouteBuilder app)
{
var g = app.MapGroup("/api").RequireAuthorization();
g.MapPost("/workflows/{id}/executions", async (
string id, ManualRunRequest req,
WorkflowRepository workflows, Executor executor, CancellationToken ct) =>
{
var wf = workflows.Get(id);
if (wf is null) return Results.NotFound();
var graph = WorkflowGraph.Parse(wf.GraphJson);
if (!graph.Nodes.Any(n => n.Id == req.StartFromNodeId))
return Results.BadRequest(new { error = "startFromNodeId is not in graph" });
var triggerItems = new Item[]
{
new(req.TriggerData ?? JsonDocument.Parse("{}").RootElement.Clone()),
};
var eid = await executor.StartExecutionAsync(
workflowId: id, graph: graph,
triggerNodeId: req.StartFromNodeId, triggerKind: "manual",
triggerItems: triggerItems, triggerData: req.TriggerData,
answerChannelId: null, ct: ct);
return Results.Accepted($"/api/executions/{eid}", new { executionId = eid });
});
g.MapGet("/workflows/{id}/executions", (
string id, ExecutionRepository ex, int? limit) =>
Results.Ok(ex.ListForWorkflow(id, Math.Clamp(limit ?? 100, 1, 500))
.Select(r => new
{
id = r.Id,
workflowId = r.WorkflowId,
trigger = r.Trigger,
status = r.Status,
startedAt = r.StartedAt,
endedAt = r.EndedAt,
error = r.Error,
})));
g.MapGet("/executions/{eid}", (
string eid, ExecutionRepository ex, NodeExecutionRepository ne) =>
{
var run = ex.Get(eid);
if (run is null) return Results.NotFound();
var nodes = ne.ListByExecution(eid).Select(n => new
{
nodeId = n.NodeId,
status = n.Status,
startedAt = n.StartedAt,
endedAt = n.EndedAt,
error = n.Error,
inputs = JsonDocument.Parse(n.InputsJson).RootElement.Clone(),
outputs = JsonDocument.Parse(n.OutputsJson).RootElement.Clone(),
});
return Results.Ok(new
{
id = run.Id,
workflowId = run.WorkflowId,
trigger = run.Trigger,
status = run.Status,
startedAt = run.StartedAt,
endedAt = run.EndedAt,
error = run.Error,
nodes,
});
});
}
}

View File

@@ -151,6 +151,7 @@ app.MapNodeTypesEndpoints();
app.MapConnectionProvidersEndpoints();
app.MapConnectionsEndpoints();
app.MapWorkflowsEndpoints();
app.MapExecutionsEndpoints();
app.MapWebhookEndpoint(TimeSpan.FromSeconds(settings.WebhookDefaultTimeoutSeconds));
app.MapGet("/api/health", () => Results.Ok(new { ok = true }));

View File

@@ -0,0 +1,66 @@
using System.Net;
using System.Net.Http.Json;
using System.Text.Json;
namespace VolcanicCore.Tests;
public class ExecutionsEndpointsTests : IClassFixture<WebFactory>
{
private readonly WebFactory _factory;
public ExecutionsEndpointsTests(WebFactory f) => _factory = f;
private async Task<HttpClient> LoggedIn()
{
var c = _factory.CreateClient(new() { HandleCookies = true, AllowAutoRedirect = false });
await c.PostAsJsonAsync("/api/session", new { password = _factory.AdminPassword });
return c;
}
[Fact(Skip = "Enabled after Tasks 22-23 land WebhookNode and WebResponseNode, and Task 24 DI-registers them")]
public async Task Manual_run_of_workflow_with_single_webhook_trigger_produces_completed_execution()
{
var c = await LoggedIn();
var graph = new
{
nodes = new object[]
{
new { id = "t", type_id = "http.webhook", config = new { path = "manual" } },
new { id = "r", type_id = "http.response", config = new { status = 204, body = "" } },
},
edges = new[] { new { from_node = "t", from_port = 0, to_node = "r", to_port = 0 } },
};
var create = await c.PostAsJsonAsync("/api/workflows", new { name = "manual-wf", graph });
var wf = await create.Content.ReadFromJsonAsync<JsonElement>();
var wfId = wf.GetProperty("id").GetString()!;
var run = await c.PostAsJsonAsync($"/api/workflows/{wfId}/executions",
new { startFromNodeId = "t", triggerData = new { hello = "world" } });
Assert.Equal(HttpStatusCode.Accepted, run.StatusCode);
var runBody = await run.Content.ReadFromJsonAsync<JsonElement>();
var eid = runBody.GetProperty("executionId").GetString()!;
// Poll for completion (executor is in-process; this returns nearly synchronously).
JsonElement detail = default;
for (var i = 0; i < 20; i++)
{
detail = await c.GetFromJsonAsync<JsonElement>($"/api/executions/{eid}");
if (detail.GetProperty("status").GetString() != "running") break;
await Task.Delay(50);
}
Assert.Equal("completed", detail.GetProperty("status").GetString());
var nodes = detail.GetProperty("nodes");
Assert.Equal(2, nodes.GetArrayLength());
var list = await c.GetFromJsonAsync<JsonElement>($"/api/workflows/{wfId}/executions");
Assert.True(list.GetArrayLength() >= 1);
}
[Fact]
public async Task Manual_run_returns_404_for_unknown_workflow()
{
var c = await LoggedIn();
var res = await c.PostAsJsonAsync("/api/workflows/nope/executions",
new { startFromNodeId = "t", triggerData = new { } });
Assert.Equal(HttpStatusCode.NotFound, res.StatusCode);
}
}