feat(executor): node pinned_input overrides upstream byPort on manual runs
This commit is contained in:
@@ -183,6 +183,16 @@ public sealed class Executor
|
|||||||
}
|
}
|
||||||
|
|
||||||
var byPort = BuildInputsByPort(portMap, allowFailurePlaceholders: false);
|
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<Item>[] { WrapPin(pinDoc.RootElement) };
|
||||||
|
}
|
||||||
|
|
||||||
var typ = _nodes.Get(node.TypeId);
|
var typ = _nodes.Get(node.TypeId);
|
||||||
if (typ is not IActionNode action)
|
if (typ is not IActionNode action)
|
||||||
throw new InvalidOperationException($"Node {nid} is not an action type.");
|
throw new InvalidOperationException($"Node {nid} is not an action type.");
|
||||||
|
|||||||
@@ -474,4 +474,76 @@ public class ExecutorTests
|
|||||||
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
||||||
Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
|
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());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user