From e24bdb012baa44002e5704ab899d0832a6667ed8 Mon Sep 17 00:00:00 2001
From: Claude <claude@volcanic>
Date: Sun, 30 Aug 2026 14:17:29 +0000
Subject: [PATCH] feat(executor): trigger pinned_output overrides triggerItems
 on manual runs

---
 src/VolcanicCore.Api/Engine/Executor.cs   | 14 +++++-
 tests/VolcanicCore.Tests/ExecutorTests.cs | 58 +++++++++++++++++++++++
 2 files changed, 70 insertions(+), 2 deletions(-)

diff --git a/src/VolcanicCore.Api/Engine/Executor.cs b/src/VolcanicCore.Api/Engine/Executor.cs
index e2d5c07..f268a91 100644
--- a/src/VolcanicCore.Api/Engine/Executor.cs
+++ b/src/VolcanicCore.Api/Engine/Executor.cs
@@ -99,7 +99,7 @@ public sealed class Executor
 
         try
         {
-            await RunAsync(eid, workflowId, graph, triggerNodeId, triggerItems, answerChannelId, logger, ct);
+            await RunAsync(eid, workflowId, graph, triggerNodeId, triggerKind, triggerItems, answerChannelId, logger, ct);
         }
         catch (Exception ex)
         {
@@ -113,11 +113,21 @@ public sealed class Executor
 
     private async Task RunAsync(
         string executionId, string workflowId, WorkflowGraph graph,
-        string triggerNodeId, IReadOnlyList<Item> triggerItems,
+        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: "[]",
diff --git a/tests/VolcanicCore.Tests/ExecutorTests.cs b/tests/VolcanicCore.Tests/ExecutorTests.cs
index d41c045..25673bd 100644
--- a/tests/VolcanicCore.Tests/ExecutorTests.cs
+++ b/tests/VolcanicCore.Tests/ExecutorTests.cs
@@ -416,4 +416,62 @@ public class ExecutorTests
         Assert.Single(items);
         Assert.Equal("42", items[0].Json.GetRawText());
     }
+
+    // --- trigger pinned_output on manual runs ------------------------------
+
+    [Fact]
+    public async Task Trigger_pinned_output_overrides_trigger_items_on_manual_run()
+    {
+        var rec = new Recorder { Id = "sink.trig_pin_manual" };
+        var (er, nr, exec, _) = Build(new TestTrigger(), rec);
+
+        var graph = new WorkflowGraph
+        {
+            Nodes = new[]
+            {
+                new WorkflowNode("t", "test.trigger", null, "{}",
+                    PinnedOutputJson: """{"pinned":true}"""),
+                new WorkflowNode("s", "sink.trig_pin_manual", null, "{}"),
+            },
+            Edges = new[] { new WorkflowEdge("t", 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);
+        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());
+    }
+
+    [Fact]
+    public async Task Trigger_pinned_output_is_ignored_on_non_manual_run()
+    {
+        var rec = new Recorder { Id = "sink.trig_pin_webhook" };
+        var (er, nr, exec, _) = Build(new TestTrigger(), rec);
+
+        var graph = new WorkflowGraph
+        {
+            Nodes = new[]
+            {
+                new WorkflowNode("t", "test.trigger", null, "{}",
+                    PinnedOutputJson: """{"pinned":true}"""),
+                new WorkflowNode("s", "sink.trig_pin_webhook", null, "{}"),
+            },
+            Edges = new[] { new WorkflowEdge("t", 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");
+        Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
+    }
 }