feat(engine): concrete ExecutionContext + TriggerActivationSink

src/VolcanicCore.Api/Engine/ExecutionContext.cs

@@ -0,0 +1,62 @@
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+using VolcanicCore.Core;
+
+namespace VolcanicCore.Api.Engine;
+
+// Interface pulled out so tests can substitute without spinning up real providers.
+// Concrete impl lives in ConnectionInstanceRegistry (next task).
+public interface IConnectionInstanceRegistry
+{
+    Task<object> GetAsync(string connectionId, CancellationToken ct);
+}
+
+public sealed class ExecutionContext : INodeContext
+{
+    private readonly IConnectionInstanceRegistry _instances;
+    private readonly string? _connectionId;
+
+    public ExecutionContext(
+        string executionId,
+        string workflowId,
+        string nodeId,
+        JsonElement config,
+        Guid? answerChannelId,
+        ILogger logger,
+        string? connectionId,
+        IConnectionInstanceRegistry instances)
+    {
+        ExecutionId = executionId;
+        WorkflowId = workflowId;
+        NodeId = nodeId;
+        Config = config;
+        AnswerChannelId = answerChannelId;
+        Logger = logger;
+        _connectionId = connectionId;
+        _instances = instances;
+    }
+
+    public string ExecutionId { get; }
+    public string WorkflowId { get; }
+    public string NodeId { get; }
+    public JsonElement Config { get; }
+    public Guid? AnswerChannelId { get; }
+    public ILogger Logger { get; }
+
+    public T GetConnection<T>() where T : class
+    {
+        if (_connectionId is null)
+            throw new InvalidOperationException(
+                $"Node {NodeId} has no configured connection but requested one of type {typeof(T).Name}.");
+
+        // Async->sync inside a node call site is acceptable here — node methods
+        // are already Task-returning and the caller is fine to block briefly for a
+        // cached lookup. Real providers should complete synchronously for a cached hit.
+        var handle = _instances.GetAsync(_connectionId, CancellationToken.None)
+            .GetAwaiter().GetResult();
+        if (handle is not T typed)
+            throw new InvalidCastException(
+                $"Connection {_connectionId} is not assignable to {typeof(T).Name}.");
+        return typed;
+    }
+}

src/VolcanicCore.Api/Engine/TriggerActivationSink.cs

@@ -0,0 +1,49 @@
+using System.Text.Json;
+using VolcanicCore.Core;
+
+namespace VolcanicCore.Api.Engine;
+
+// Callback shape the TriggerHost hands to a trigger's StartAsync. Emitting
+// via the sink kicks off one execution.
+public delegate Task StartExecutionAsync(
+    string workflowId,
+    string triggerNodeId,
+    string triggerKind,             // "nats.hook" | "http.webhook" | ...
+    IReadOnlyList<Item> items,
+    JsonElement? triggerData,
+    Guid? answerChannelId,
+    CancellationToken ct);
+
+public sealed class TriggerActivationSink : ITriggerSink
+{
+    private readonly string _workflowId;
+    private readonly string _triggerNodeId;
+    private readonly string _triggerKind;
+    private readonly StartExecutionAsync _start;
+
+    public TriggerActivationSink(
+        string workflowId, string triggerNodeId, string triggerKind, StartExecutionAsync start)
+    {
+        _workflowId = workflowId;
+        _triggerNodeId = triggerNodeId;
+        _triggerKind = triggerKind;
+        _start = start;
+    }
+
+    public Task EmitAsync(
+        IReadOnlyList<Item> items,
+        JsonElement? triggerData = null,
+        Guid? answerChannelId = null,
+        CancellationToken ct = default)
+    {
+        // Fire-and-forget: a NATS trigger's pump must not block per-event on
+        // execution completion. A Webhook trigger's /wh endpoint doesn't rely
+        // on this task either — it awaits WebhookRouter.Register for the reply.
+        _ = Task.Run(async () =>
+        {
+            try { await _start(_workflowId, _triggerNodeId, _triggerKind, items, triggerData, answerChannelId, ct); }
+            catch { /* Executor already logged + persisted the failure */ }
+        });
+        return Task.CompletedTask;
+    }
+}

tests/VolcanicCore.Tests/ExecutionContextTests.cs

@@ -0,0 +1,59 @@
+using System.Text.Json;
+using Microsoft.Extensions.Logging.Abstractions;
+using VolcanicCore.Api.Engine;
+using ExecutionContext = VolcanicCore.Api.Engine.ExecutionContext;
+
+namespace VolcanicCore.Tests;
+
+public class ExecutionContextTests
+{
+    private sealed class Handle { public string Value { get; set; } = ""; }
+
+    [Fact]
+    public void GetConnection_returns_registered_handle_by_id()
+    {
+        var handle = new Handle { Value = "live" };
+        var reg = FakeInstanceRegistry.With("conn-1", handle);
+        var ctx = new ExecutionContext(
+            executionId: "e",
+            workflowId: "w",
+            nodeId: "n",
+            config: JsonDocument.Parse("{}").RootElement,
+            answerChannelId: null,
+            logger: NullLogger.Instance,
+            connectionId: "conn-1",
+            instances: reg);
+
+        Assert.Same(handle, ctx.GetConnection<Handle>());
+    }
+
+    [Fact]
+    public void GetConnection_throws_when_node_has_no_connection_id()
+    {
+        var ctx = new ExecutionContext("e", "w", "n",
+            JsonDocument.Parse("{}").RootElement, null,
+            NullLogger.Instance, connectionId: null, instances: FakeInstanceRegistry.Empty);
+
+        Assert.Throws<InvalidOperationException>(() => ctx.GetConnection<Handle>());
+    }
+
+    [Fact]
+    public void GetConnection_throws_when_type_mismatches()
+    {
+        var reg = FakeInstanceRegistry.With("conn-1", new object());
+        var ctx = new ExecutionContext("e", "w", "n",
+            JsonDocument.Parse("{}").RootElement, null,
+            NullLogger.Instance, connectionId: "conn-1", instances: reg);
+
+        Assert.Throws<InvalidCastException>(() => ctx.GetConnection<Handle>());
+    }
+
+    private sealed class FakeInstanceRegistry : IConnectionInstanceRegistry
+    {
+        private readonly Dictionary<string, object> _byId;
+        private FakeInstanceRegistry(Dictionary<string, object> byId) => _byId = byId;
+        public static FakeInstanceRegistry Empty => new(new());
+        public static FakeInstanceRegistry With(string id, object h) => new(new() { [id] = h });
+        public Task<object> GetAsync(string connectionId, CancellationToken ct) => Task.FromResult(_byId[connectionId]);
+    }
+}