diff --git a/src/VolcanicCore.Api/Engine/Executor.cs b/src/VolcanicCore.Api/Engine/Executor.cs index f268a91..386f914 100644 --- a/src/VolcanicCore.Api/Engine/Executor.cs +++ b/src/VolcanicCore.Api/Engine/Executor.cs @@ -183,6 +183,16 @@ public sealed class Executor } var byPort = BuildInputsByPort(portMap, allowFailurePlaceholders: false); + + // Manual runs honor the node's pinned_input — replaces byPort[0] with the + // pinned items. Node's own logic still runs (we intentionally don't skip + // upstream ancestors; their output is discarded for THIS node's byPort). + if (triggerKind == "manual" && node.PinnedInputJson is { } nodePinJson) + { + using var pinDoc = JsonDocument.Parse(nodePinJson); + byPort = new IReadOnlyList[] { WrapPin(pinDoc.RootElement) }; + } + var typ = _nodes.Get(node.TypeId); if (typ is not IActionNode action) throw new InvalidOperationException($"Node {nid} is not an action type."); diff --git a/tests/VolcanicCore.Tests/ExecutorTests.cs b/tests/VolcanicCore.Tests/ExecutorTests.cs index 25673bd..3dcbae7 100644 --- a/tests/VolcanicCore.Tests/ExecutorTests.cs +++ b/tests/VolcanicCore.Tests/ExecutorTests.cs @@ -474,4 +474,76 @@ public class ExecutorTests var port0Item0Json = inputs.RootElement[0][0].GetProperty("json"); Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean()); } + + // --- node pinned_input on manual runs --------------------------------- + + [Fact] + public async Task Node_pinned_input_overrides_upstream_output_on_manual_run() + { + var rec = new Recorder { Id = "sink.node_pin_manual" }; + var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec); + + var graph = new WorkflowGraph + { + Nodes = new[] + { + new WorkflowNode("t", "test.trigger", null, "{}"), + new WorkflowNode("a", "test.echo", null, "{}"), + new WorkflowNode("s", "sink.node_pin_manual", null, "{}", + PinnedInputJson: """{"pinned":true}"""), + }, + Edges = new[] + { + new WorkflowEdge("t", 0, "a", 0), + new WorkflowEdge("a", 0, "s", 0), + }, + }; + + var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false })); + var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual", + new[] { passedIn }, null, null, CancellationToken.None); + + Assert.Equal("completed", er.Get(eid)!.Status); + // Sink's persisted inputs must be the pin, not what `a` emitted. + var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s"); + using var inputs = JsonDocument.Parse(sinkRow.InputsJson); + var port0Item0Json = inputs.RootElement[0][0].GetProperty("json"); + Assert.True(port0Item0Json.GetProperty("pinned").GetBoolean()); + // `a` still ran and its own output row exists — confirming we don't skip upstream. + Assert.Contains(nr.ListByExecution(eid), r => r.NodeId == "a" && r.Status == "completed"); + } + + [Fact] + public async Task Node_pinned_input_is_ignored_on_non_manual_run() + { + var rec = new Recorder { Id = "sink.node_pin_webhook" }; + var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec); + + var graph = new WorkflowGraph + { + Nodes = new[] + { + new WorkflowNode("t", "test.trigger", null, "{}"), + new WorkflowNode("a", "test.echo", null, "{}"), + new WorkflowNode("s", "sink.node_pin_webhook", null, "{}", + PinnedInputJson: """{"pinned":true}"""), + }, + Edges = new[] + { + new WorkflowEdge("t", 0, "a", 0), + new WorkflowEdge("a", 0, "s", 0), + }, + }; + + var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false })); + var eid = await exec.StartExecutionAsync("wf", graph, "t", "http.webhook", + new[] { passedIn }, null, null, CancellationToken.None); + + Assert.Equal("completed", er.Get(eid)!.Status); + var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s"); + using var inputs = JsonDocument.Parse(sinkRow.InputsJson); + var port0Item0Json = inputs.RootElement[0][0].GetProperty("json"); + // Real upstream (echo of the trigger's `pinned:false`) reaches the sink. + Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean()); + } }