feat(engine): TriggerHost with activation/deactivation + startup abort sweep

src/VolcanicCore.Api/Endpoints/WorkflowsEndpoints.cs

@@ -1,5 +1,6 @@
 using System.Text.Json;
 using VolcanicCore.Api.Contracts;
+using VolcanicCore.Api.Engine;
 using VolcanicCore.Persistence.Repositories;
 
 namespace VolcanicCore.Api.Endpoints;
@@ -29,23 +30,45 @@ public static class WorkflowsEndpoints
             return Results.Created($"/api/workflows/{id}", ToDto(w));
         });
 
-        g.MapPut("/{id}", (string id, UpdateWorkflowRequest req, WorkflowRepository repo) =>
+        g.MapPut("/{id}", async (string id, UpdateWorkflowRequest req,
+            WorkflowRepository repo, IWorkflowLifecycle host, CancellationToken ct) =>
         {
             if (req.Graph.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null)
                 return Results.BadRequest(new { error = "graph is required" });
 
-            return repo.Update(id, req.Name, req.Graph.GetRawText())
-                ? Results.NoContent() : Results.NotFound();
+            var current = repo.Get(id);
+            if (current is null) return Results.NotFound();
+            var wasActive = current.Active;
+
+            if (!repo.Update(id, req.Name, req.Graph.GetRawText())) return Results.NotFound();
+
+            if (wasActive)
+            {
+                await host.StopAsync(id);
+                await host.StartAsync(id, ct);
+            }
+            return Results.NoContent();
         });
 
-        g.MapPost("/{id}/activate", (string id, WorkflowRepository repo) =>
-            repo.SetActive(id, true) ? Results.NoContent() : Results.NotFound());
+        g.MapPost("/{id}/activate", async (string id, WorkflowRepository repo, IWorkflowLifecycle host, CancellationToken ct) =>
+        {
+            if (!repo.SetActive(id, true)) return Results.NotFound();
+            await host.StartAsync(id, ct);
+            return Results.NoContent();
+        });
 
-        g.MapPost("/{id}/deactivate", (string id, WorkflowRepository repo) =>
-            repo.SetActive(id, false) ? Results.NoContent() : Results.NotFound());
+        g.MapPost("/{id}/deactivate", async (string id, WorkflowRepository repo, IWorkflowLifecycle host) =>
+        {
+            if (!repo.SetActive(id, false)) return Results.NotFound();
+            await host.StopAsync(id);
+            return Results.NoContent();
+        });
 
-        g.MapDelete("/{id}", (string id, WorkflowRepository repo) =>
-            repo.Delete(id) ? Results.NoContent() : Results.NotFound());
+        g.MapDelete("/{id}", async (string id, WorkflowRepository repo, IWorkflowLifecycle host) =>
+        {
+            await host.StopAsync(id);
+            return repo.Delete(id) ? Results.NoContent() : Results.NotFound();
+        });
     }
 
     private static WorkflowResponse ToDto(WorkflowRow w)

src/VolcanicCore.Api/Engine/IWorkflowLifecycle.cs

@@ -0,0 +1,7 @@
+namespace VolcanicCore.Api.Engine;
+
+public interface IWorkflowLifecycle
+{
+    Task StartAsync(string workflowId, CancellationToken ct);
+    Task StopAsync(string workflowId);
+}

src/VolcanicCore.Api/Engine/TriggerHost.cs

@@ -0,0 +1,114 @@
+using System.Text.Json;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using VolcanicCore.Api.Registries;
+using VolcanicCore.Core;
+using VolcanicCore.Persistence.Repositories;
+
+namespace VolcanicCore.Api.Engine;
+
+public sealed class TriggerHost : IHostedService, IWorkflowLifecycle, IAsyncDisposable
+{
+    private readonly WorkflowRepository _workflows;
+    private readonly ExecutionRepository _executions;
+    private readonly NodeRegistry _nodes;
+    private readonly IConnectionInstanceRegistry _instances;
+    private readonly Executor _executor;
+    private readonly ILoggerFactory _lf;
+    private readonly Dictionary<string, List<IAsyncDisposable>> _active = new(StringComparer.Ordinal);
+    private readonly SemaphoreSlim _mu = new(1, 1);
+
+    public TriggerHost(
+        WorkflowRepository workflows, ExecutionRepository executions,
+        NodeRegistry nodes, IConnectionInstanceRegistry instances,
+        Executor executor, ILoggerFactory lf)
+    {
+        _workflows = workflows;
+        _executions = executions;
+        _nodes = nodes;
+        _instances = instances;
+        _executor = executor;
+        _lf = lf;
+    }
+
+    public async Task StartAsync(CancellationToken ct)
+    {
+        RunStartupAbortSweep();
+        foreach (var w in _workflows.List().Where(w => w.Active))
+            await StartAsync(w.Id, ct);
+    }
+
+    // Exposed so tests can invoke it deterministically.
+    public void RunStartupAbortSweep() => _executions.AbortAllRunning();
+
+    public async Task StartAsync(string workflowId, CancellationToken ct)
+    {
+        await _mu.WaitAsync(ct);
+        try
+        {
+            if (_active.ContainsKey(workflowId)) return;
+            var row = _workflows.Get(workflowId);
+            if (row is null) return;
+
+            var graph = WorkflowGraph.Parse(row.GraphJson);
+            var disposables = new List<IAsyncDisposable>();
+            foreach (var node in graph.Nodes)
+            {
+                var typ = _nodes.Get(node.TypeId);
+                if (typ is not ITriggerNode trigger) continue;
+
+                using var cfgDoc = JsonDocument.Parse(node.ConfigJson);
+                var ctx = new ExecutionContext(
+                    executionId: "",           // no execution yet — trigger is activating
+                    workflowId: workflowId,
+                    nodeId: node.Id,
+                    config: cfgDoc.RootElement.Clone(),
+                    answerChannelId: null,
+                    logger: _lf.CreateLogger($"trigger.{node.Id}"),
+                    connectionId: node.ConnectionId,
+                    instances: _instances);
+
+                var sink = new TriggerActivationSink(workflowId, node.Id, trigger.Id,
+                    (wf, tn, kind, items, td, ans, ictCt) =>
+                        _executor.StartExecutionAsync(wf, graph, tn, kind, items, td, ans, ictCt));
+
+                var handle = await trigger.StartAsync(ctx, sink, ct);
+                disposables.Add(handle);
+            }
+            _active[workflowId] = disposables;
+        }
+        finally { _mu.Release(); }
+    }
+
+    public async Task StopAsync(string workflowId)
+    {
+        List<IAsyncDisposable>? disp = null;
+        await _mu.WaitAsync();
+        try
+        {
+            if (_active.Remove(workflowId, out var handles)) disp = handles;
+        }
+        finally { _mu.Release(); }
+
+        if (disp is null) return;
+        foreach (var d in disp) try { await d.DisposeAsync(); } catch { }
+    }
+
+    public async Task StopAsync(CancellationToken ct) => await DisposeAsync();
+
+    public async ValueTask DisposeAsync()
+    {
+        List<List<IAsyncDisposable>> snapshot;
+        await _mu.WaitAsync();
+        try
+        {
+            snapshot = _active.Values.ToList();
+            _active.Clear();
+        }
+        finally { _mu.Release(); }
+
+        foreach (var handles in snapshot)
+            foreach (var d in handles)
+                try { await d.DisposeAsync(); } catch { }
+    }
+}

src/VolcanicCore.Api/Program.cs

@@ -120,6 +120,11 @@ builder.Services.AddSingleton<IConnectionInstanceRegistry>(sp =>
 builder.Services.AddSingleton<WebhookRouter>();
 builder.Services.AddSingleton<WebhookRegistry>();
 
+builder.Services.AddSingleton<Executor>();
+builder.Services.AddSingleton<TriggerHost>();
+builder.Services.AddSingleton<IWorkflowLifecycle>(sp => sp.GetRequiredService<TriggerHost>());
+builder.Services.AddHostedService(sp => sp.GetRequiredService<TriggerHost>());
+
 var app = builder.Build();
 
 // Apply migrations at boot.