Orchestrator decomposition: replace JobEngine with release-orchestrator + workflow services

- Remove jobengine and jobengine-worker containers from docker-compose
- Create release-orchestrator service (120 endpoints) with full auth, tenant, and infrastructure DI
- Wire workflow engine to PostgreSQL with definition store (wf_definitions table)
- Deploy 4 canonical workflow definitions on startup (release-promotion, scan-execution, advisory-refresh, compliance-sweep)
- Fix workflow definition JSON to match canonical contract schema (set-state, call-transport, decision)
- Add WorkflowClient to release-orchestrator for starting workflow instances on promotion
- Add WorkflowTriggerClient + endpoint to scheduler for triggering workflows from system schedules
- Update gateway routes from jobengine.stella-ops.local to release-orchestrator.stella-ops.local
- Remove Platform.Database dependency on JobEngine.Infrastructure
- Fix workflow csproj duplicate Content items (EmbeddedResource + SDK default)
- System-managed schedules with source column, SystemScheduleBootstrap, inline edit UI

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/Workflow/StellaOps.Workflow.WebService/Bootstrap/WorkflowDefinitionBootstrap.cs

@@ -0,0 +1,122 @@
+using System.Reflection;
+using StellaOps.Workflow.Contracts;
+using StellaOps.Workflow.Engine.Services;
+
+namespace StellaOps.Workflow.WebService.Bootstrap;
+
+/// <summary>
+/// Deploys built-in workflow definitions on startup.
+/// Definitions are embedded as JSON resources in the Definitions/ folder.
+/// Import is idempotent — existing definitions with matching content hash are skipped.
+/// </summary>
+public sealed class WorkflowDefinitionBootstrap : BackgroundService
+{
+    private readonly IServiceScopeFactory _scopeFactory;
+    private readonly ILogger<WorkflowDefinitionBootstrap> _logger;
+
+    private static readonly string[] DefinitionNames =
+    [
+        "release-promotion",
+        "scan-execution",
+        "advisory-refresh",
+        "compliance-sweep",
+    ];
+
+    public WorkflowDefinitionBootstrap(
+        IServiceScopeFactory scopeFactory,
+        ILogger<WorkflowDefinitionBootstrap> logger)
+    {
+        _scopeFactory = scopeFactory;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        // Wait briefly for the database and other services to initialize
+        await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
+
+        try
+        {
+            await using var scope = _scopeFactory.CreateAsyncScope();
+            var deploymentApi = scope.ServiceProvider.GetRequiredService<WorkflowDefinitionDeploymentService>();
+
+            foreach (var name in DefinitionNames)
+            {
+                if (stoppingToken.IsCancellationRequested) break;
+
+                try
+                {
+                    var json = await LoadDefinitionJsonAsync(name);
+                    if (string.IsNullOrWhiteSpace(json))
+                    {
+                        _logger.LogWarning("Workflow definition resource not found: {Name}", name);
+                        continue;
+                    }
+
+                    var response = await deploymentApi.ImportAsync(new WorkflowDefinitionImportRequest
+                    {
+                        CanonicalDefinitionJson = json,
+                        ImportedBy = "system-bootstrap",
+                    }, stoppingToken);
+
+                    if (response.WasImported)
+                    {
+                        _logger.LogInformation(
+                            "Deployed workflow definition {Name} v{Version} (hash: {Hash})",
+                            response.WorkflowName, response.Version, response.ContentHash);
+                    }
+                    else if (response.ValidationIssues is { Count: > 0 })
+                    {
+                        _logger.LogWarning(
+                            "Workflow definition {Name} failed validation: {Issues}",
+                            response.WorkflowName,
+                            string.Join("; ", response.ValidationIssues));
+                    }
+                    else
+                    {
+                        _logger.LogInformation(
+                            "Workflow definition {Name} already up-to-date (hash: {Hash})",
+                            response.WorkflowName, response.ContentHash);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex, "Failed to deploy workflow definition: {Name}", name);
+                }
+            }
+
+            _logger.LogInformation("Workflow definition bootstrap complete ({Count} definitions processed)", DefinitionNames.Length);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Workflow definition bootstrap failed");
+        }
+    }
+
+    private static async Task<string?> LoadDefinitionJsonAsync(string name)
+    {
+        // Try embedded resource first
+        var assembly = Assembly.GetExecutingAssembly();
+        var resourceName = assembly.GetManifestResourceNames()
+            .FirstOrDefault(n => n.EndsWith($"{name}.json", StringComparison.OrdinalIgnoreCase));
+
+        if (resourceName is not null)
+        {
+            await using var stream = assembly.GetManifestResourceStream(resourceName);
+            if (stream is not null)
+            {
+                using var reader = new StreamReader(stream);
+                return await reader.ReadToEndAsync();
+            }
+        }
+
+        // Fallback: load from file system (development)
+        var filePath = Path.Combine(AppContext.BaseDirectory, "Definitions", $"{name}.json");
+        if (!File.Exists(filePath))
+        {
+            filePath = Path.Combine("Definitions", $"{name}.json");
+        }
+
+        return File.Exists(filePath) ? await File.ReadAllTextAsync(filePath) : null;
+    }
+}

src/Workflow/StellaOps.Workflow.WebService/Definitions/advisory-refresh.json

@@ -0,0 +1,56 @@
+{
+  "schemaVersion": "stellaops.workflow.definition/v1",
+  "workflowName": "advisory-refresh",
+  "workflowVersion": "1.0.0",
+  "displayName": "Advisory Refresh",
+  "startRequest": {
+    "contractName": "StellaOps.Concelier.Contracts.AdvisoryRefreshRequest",
+    "schema": {
+      "type": "object",
+      "properties": {
+        "feedIds": { "type": "array", "items": { "type": "string" } }
+      }
+    },
+    "allowAdditionalProperties": true
+  },
+  "workflowRoles": [],
+  "start": {
+    "initializeStateExpression": {
+      "$type": "object",
+      "properties": [
+        { "name": "feedIds", "expression": { "$type": "path", "path": "start.feedIds" } },
+        { "name": "status", "expression": { "$type": "string", "value": "refreshing" } }
+      ]
+    },
+    "initialSequence": {
+      "steps": [
+        {
+          "$type": "call-transport",
+          "stepName": "refresh-advisory-feeds",
+          "invocation": {
+            "address": {
+              "$type": "microservice",
+              "microserviceName": "concelier",
+              "command": "refresh-feeds"
+            },
+            "payloadExpression": {
+              "$type": "object",
+              "properties": [
+                { "name": "feedIds", "expression": { "$type": "path", "path": "state.feedIds" } }
+              ]
+            }
+          },
+          "resultKey": "refreshResult"
+        },
+        {
+          "$type": "set-state",
+          "stateKey": "status",
+          "valueExpression": { "$type": "string", "value": "completed" }
+        },
+        { "$type": "complete" }
+      ]
+    }
+  },
+  "tasks": [],
+  "requiredModules": []
+}

src/Workflow/StellaOps.Workflow.WebService/Definitions/compliance-sweep.json

@@ -0,0 +1,59 @@
+{
+  "schemaVersion": "stellaops.workflow.definition/v1",
+  "workflowName": "compliance-sweep",
+  "workflowVersion": "1.0.0",
+  "displayName": "Compliance Sweep",
+  "startRequest": {
+    "contractName": "StellaOps.Policy.Contracts.ComplianceSweepRequest",
+    "schema": {
+      "type": "object",
+      "properties": {
+        "policySetId": { "type": "string" },
+        "scope": { "type": "string" }
+      }
+    },
+    "allowAdditionalProperties": true
+  },
+  "workflowRoles": [],
+  "start": {
+    "initializeStateExpression": {
+      "$type": "object",
+      "properties": [
+        { "name": "policySetId", "expression": { "$type": "path", "path": "start.policySetId" } },
+        { "name": "scope", "expression": { "$type": "path", "path": "start.scope" } },
+        { "name": "status", "expression": { "$type": "string", "value": "sweeping" } }
+      ]
+    },
+    "initialSequence": {
+      "steps": [
+        {
+          "$type": "call-transport",
+          "stepName": "evaluate-compliance",
+          "invocation": {
+            "address": {
+              "$type": "microservice",
+              "microserviceName": "policy-engine",
+              "command": "evaluate-compliance-sweep"
+            },
+            "payloadExpression": {
+              "$type": "object",
+              "properties": [
+                { "name": "policySetId", "expression": { "$type": "path", "path": "state.policySetId" } },
+                { "name": "scope", "expression": { "$type": "path", "path": "state.scope" } }
+              ]
+            }
+          },
+          "resultKey": "complianceResult"
+        },
+        {
+          "$type": "set-state",
+          "stateKey": "status",
+          "valueExpression": { "$type": "string", "value": "completed" }
+        },
+        { "$type": "complete" }
+      ]
+    }
+  },
+  "tasks": [],
+  "requiredModules": []
+}

src/Workflow/StellaOps.Workflow.WebService/Definitions/release-promotion.json

@@ -0,0 +1,167 @@
+{
+  "schemaVersion": "stellaops.workflow.definition/v1",
+  "workflowName": "release-promotion",
+  "workflowVersion": "1.0.0",
+  "displayName": "Release Promotion",
+  "startRequest": {
+    "contractName": "StellaOps.ReleaseOrchestrator.Contracts.ReleasePromotionRequest",
+    "schema": {
+      "type": "object",
+      "properties": {
+        "releaseId": { "type": "string" },
+        "targetEnvironment": { "type": "string" },
+        "requestedBy": { "type": "string" }
+      },
+      "required": ["releaseId", "targetEnvironment", "requestedBy"]
+    },
+    "allowAdditionalProperties": true
+  },
+  "workflowRoles": ["release-approver", "release-operator"],
+  "businessReference": {
+    "keyExpression": { "$type": "path", "path": "start.releaseId" },
+    "parts": [
+      { "name": "releaseId", "expression": { "$type": "path", "path": "start.releaseId" } },
+      { "name": "environment", "expression": { "$type": "path", "path": "start.targetEnvironment" } }
+    ]
+  },
+  "start": {
+    "initializeStateExpression": {
+      "$type": "object",
+      "properties": [
+        { "name": "releaseId", "expression": { "$type": "path", "path": "start.releaseId" } },
+        { "name": "targetEnvironment", "expression": { "$type": "path", "path": "start.targetEnvironment" } },
+        { "name": "requestedBy", "expression": { "$type": "path", "path": "start.requestedBy" } },
+        { "name": "outcome", "expression": { "$type": "string", "value": "pending" } }
+      ]
+    },
+    "initialSequence": {
+      "steps": [
+        {
+          "$type": "set-state",
+          "stateKey": "outcome",
+          "valueExpression": { "$type": "string", "value": "evaluating-gates" }
+        },
+        {
+          "$type": "call-transport",
+          "stepName": "evaluate-security-gates",
+          "invocation": {
+            "address": {
+              "$type": "microservice",
+              "microserviceName": "policy-engine",
+              "command": "evaluate-release-gates"
+            },
+            "payloadExpression": {
+              "$type": "object",
+              "properties": [
+                { "name": "releaseId", "expression": { "$type": "path", "path": "state.releaseId" } },
+                { "name": "environment", "expression": { "$type": "path", "path": "state.targetEnvironment" } }
+              ]
+            }
+          },
+          "resultKey": "gateResult"
+        },
+        {
+          "$type": "decision",
+          "decisionName": "gates-passed-check",
+          "conditionExpression": {
+            "$type": "binary",
+            "operator": "==",
+            "left": { "$type": "path", "path": "state.gateResult.passed" },
+            "right": { "$type": "boolean", "value": true }
+          },
+          "whenTrue": {
+            "steps": [
+              {
+                "$type": "set-state",
+                "stateKey": "outcome",
+                "valueExpression": { "$type": "string", "value": "awaiting-approval" }
+              },
+              {
+                "$type": "activate-task",
+                "taskName": "Approve Release Promotion"
+              }
+            ]
+          },
+          "whenElse": {
+            "steps": [
+              {
+                "$type": "set-state",
+                "stateKey": "outcome",
+                "valueExpression": { "$type": "string", "value": "gates-failed" }
+              },
+              { "$type": "complete" }
+            ]
+          }
+        }
+      ]
+    }
+  },
+  "tasks": [
+    {
+      "taskName": "Approve Release Promotion",
+      "taskType": "approval",
+      "routeExpression": { "$type": "string", "value": "release-approval" },
+      "payloadExpression": { "$type": "path", "path": "state" },
+      "taskRoles": ["release-approver"],
+      "onComplete": {
+        "steps": [
+          {
+            "$type": "decision",
+            "decisionName": "approval-decision",
+            "conditionExpression": {
+              "$type": "binary",
+              "operator": "==",
+              "left": { "$type": "path", "path": "task.result.decision" },
+              "right": { "$type": "string", "value": "approved" }
+            },
+            "whenTrue": {
+              "steps": [
+                {
+                  "$type": "set-state",
+                  "stateKey": "outcome",
+                  "valueExpression": { "$type": "string", "value": "deploying" }
+                },
+                {
+                  "$type": "call-transport",
+                  "stepName": "execute-deployment",
+                  "invocation": {
+                    "address": {
+                      "$type": "microservice",
+                      "microserviceName": "release-orchestrator",
+                      "command": "execute-deployment"
+                    },
+                    "payloadExpression": {
+                      "$type": "object",
+                      "properties": [
+                        { "name": "releaseId", "expression": { "$type": "path", "path": "state.releaseId" } },
+                        { "name": "environment", "expression": { "$type": "path", "path": "state.targetEnvironment" } }
+                      ]
+                    }
+                  },
+                  "resultKey": "deploymentResult"
+                },
+                {
+                  "$type": "set-state",
+                  "stateKey": "outcome",
+                  "valueExpression": { "$type": "string", "value": "completed" }
+                },
+                { "$type": "complete" }
+              ]
+            },
+            "whenElse": {
+              "steps": [
+                {
+                  "$type": "set-state",
+                  "stateKey": "outcome",
+                  "valueExpression": { "$type": "string", "value": "rejected" }
+                },
+                { "$type": "complete" }
+              ]
+            }
+          }
+        ]
+      }
+    }
+  ],
+  "requiredModules": []
+}

src/Workflow/StellaOps.Workflow.WebService/Definitions/scan-execution.json

@@ -0,0 +1,71 @@
+{
+  "schemaVersion": "stellaops.workflow.definition/v1",
+  "workflowName": "scan-execution",
+  "workflowVersion": "1.0.0",
+  "displayName": "Scan Execution",
+  "startRequest": {
+    "contractName": "StellaOps.Scanner.Contracts.ScanExecutionRequest",
+    "schema": {
+      "type": "object",
+      "properties": {
+        "scanType": { "type": "string" },
+        "targets": { "type": "array", "items": { "type": "string" } }
+      },
+      "required": ["scanType"]
+    },
+    "allowAdditionalProperties": true
+  },
+  "workflowRoles": ["scanner-operator"],
+  "businessReference": {
+    "keyExpression": { "$type": "path", "path": "instance.id" },
+    "parts": [
+      { "name": "scanType", "expression": { "$type": "path", "path": "start.scanType" } }
+    ]
+  },
+  "start": {
+    "initializeStateExpression": {
+      "$type": "object",
+      "properties": [
+        { "name": "scanType", "expression": { "$type": "path", "path": "start.scanType" } },
+        { "name": "targets", "expression": { "$type": "path", "path": "start.targets" } },
+        { "name": "status", "expression": { "$type": "string", "value": "queued" } }
+      ]
+    },
+    "initialSequence": {
+      "steps": [
+        {
+          "$type": "set-state",
+          "stateKey": "status",
+          "valueExpression": { "$type": "string", "value": "scanning" }
+        },
+        {
+          "$type": "call-transport",
+          "stepName": "run-scan",
+          "invocation": {
+            "address": {
+              "$type": "microservice",
+              "microserviceName": "scanner",
+              "command": "execute-scan"
+            },
+            "payloadExpression": {
+              "$type": "object",
+              "properties": [
+                { "name": "scanType", "expression": { "$type": "path", "path": "state.scanType" } },
+                { "name": "targets", "expression": { "$type": "path", "path": "state.targets" } }
+              ]
+            }
+          },
+          "resultKey": "scanResult"
+        },
+        {
+          "$type": "set-state",
+          "stateKey": "status",
+          "valueExpression": { "$type": "string", "value": "completed" }
+        },
+        { "$type": "complete" }
+      ]
+    }
+  },
+  "tasks": [],
+  "requiredModules": []
+}

src/Workflow/StellaOps.Workflow.WebService/Program.cs

@@ -1,8 +1,10 @@
-using StellaOps.Workflow.DataStore.MongoDB;
+using StellaOps.Workflow.DataStore.PostgreSQL;
 using StellaOps.Workflow.Engine.Authorization;
 using StellaOps.Workflow.Engine.Services;
-using StellaOps.Workflow.Signaling.Redis;
 using StellaOps.Workflow.WebService.Endpoints;
+using StellaOps.Router.AspNet;
+using StellaOps.Workflow.WebService.Bootstrap;
+using System.Reflection;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -15,21 +17,30 @@ builder.Services.AddWorkflowEngineHostedServices();
 // Authorization service (required by WorkflowRuntimeService)
 builder.Services.AddScoped<WorkflowTaskAuthorizationService>();
 
-// MongoDB data store (projection store, runtime state, signals, dead letters, etc.)
-builder.Services.AddWorkflowMongoDataStore(builder.Configuration);
+// PostgreSQL data store (projection store, runtime state, signals, dead letters, etc.)
+builder.Services.AddWorkflowPostgresDataStore(builder.Configuration);
 
-// Redis signaling driver (wake notifications across instances)
-builder.Services.AddWorkflowRedisSignaling(builder.Configuration);
+// Stella Router integration
+var routerEnabled = builder.Services.AddRouterMicroservice(
+    builder.Configuration,
+    serviceName: "workflow",
+    version: Assembly.GetExecutingAssembly()
+        .GetCustomAttribute<AssemblyInformationalVersionAttribute>()
+        ?.InformationalVersion ?? "1.0.0",
+    routerOptionsSection: "Router");
 
-// Rendering layout engines can be registered here when available:
-// builder.Services.AddWorkflowElkSharpRenderer();
-// builder.Services.AddWorkflowSvgRenderer();
+// Deploy built-in workflow definitions on startup
+builder.Services.AddHostedService<WorkflowDefinitionBootstrap>();
 
 var app = builder.Build();
 
+app.TryUseStellaRouter(routerEnabled);
+
 // Map all workflow API endpoints under /api/workflow
 app.MapWorkflowEndpoints();
 
+app.TryRefreshStellaRouterEndpoints(routerEnabled);
+
 await app.RunAsync();
 
 public partial class Program { }

src/Workflow/StellaOps.Workflow.WebService/StellaOps.Workflow.WebService.csproj

@@ -9,15 +9,21 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <EmbeddedResource Include="Definitions\*.json" />
+    <!-- SDK auto-includes Content items; rely on EmbeddedResource for definitions -->
+    <Content Update="Definitions\*.json" CopyToOutputDirectory="PreserveNewest" />
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="../../Router/__Libraries/StellaOps.Router.AspNet/StellaOps.Router.AspNet.csproj" />
+    <ProjectReference Include="../../Router/__Libraries/StellaOps.Microservice/StellaOps.Microservice.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Engine/StellaOps.Workflow.Engine.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Abstractions/StellaOps.Workflow.Abstractions.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Contracts/StellaOps.Workflow.Contracts.csproj" />
 
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.DataStore.Oracle/StellaOps.Workflow.DataStore.Oracle.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.DataStore.MongoDB/StellaOps.Workflow.DataStore.MongoDB.csproj" />
+    <ProjectReference Include="../__Libraries/StellaOps.Workflow.DataStore.PostgreSQL/StellaOps.Workflow.DataStore.PostgreSQL.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Signaling.Redis/StellaOps.Workflow.Signaling.Redis.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Renderer.Svg/StellaOps.Workflow.Renderer.Svg.csproj" />
     <ProjectReference Include="../__Libraries/StellaOps.Workflow.Renderer.ElkSharp/StellaOps.Workflow.Renderer.ElkSharp.csproj" />

src/Workflow/StellaOps.Workflow.WebService/appsettings.json

@@ -18,16 +18,16 @@
     ]
   },
   "WorkflowBackend": {
-    "Provider": "Mongo",
-    "Mongo": {
-      "ConnectionStringName": "WorkflowMongo",
-      "DatabaseName": "stellaops_workflow"
+    "Provider": "Postgres",
+    "Postgres": {
+      "ConnectionStringName": "WorkflowPostgres",
+      "SchemaName": "workflow"
     }
   },
   "WorkflowSignalDriver": {
     "Provider": "Native"
   },
   "ConnectionStrings": {
-    "WorkflowMongo": "mongodb://localhost:27017"
+    "WorkflowPostgres": "Host=db.stella-ops.local;Port=5432;Database=stellaops_platform;Username=stellaops;Password=stellaops;Maximum Pool Size=20;Minimum Pool Size=2"
   }
 }

src/Workflow/__Libraries/StellaOps.Workflow.DataStore.PostgreSQL/PostgresWorkflowDataStoreExtensions.cs

@@ -43,6 +43,7 @@ public static class PostgresWorkflowDataStoreExtensions
         services.Replace(ServiceDescriptor.Scoped<IWorkflowProjectionRetentionStore, PostgresWorkflowProjectionRetentionStore>());
         services.Replace(ServiceDescriptor.Scoped<IWorkflowMutationCoordinator, PostgresWorkflowMutationCoordinator>());
         services.Replace(ServiceDescriptor.Scoped<IWorkflowProjectionStore, PostgresWorkflowProjectionStore>());
+        services.Replace(ServiceDescriptor.Scoped<IWorkflowDefinitionStore, PostgresWorkflowDefinitionStore>());
         services.Replace(ServiceDescriptor.Scoped<IWorkflowSignalStore>(sp => sp.GetRequiredService<PostgresWorkflowSignalStore>()));
         services.Replace(ServiceDescriptor.Scoped<IWorkflowSignalClaimStore>(sp => sp.GetRequiredService<PostgresWorkflowSignalStore>()));
         services.Replace(ServiceDescriptor.Scoped<IWorkflowSignalScheduler>(sp => sp.GetRequiredService<PostgresWorkflowScheduleBus>()));

src/Workflow/__Libraries/StellaOps.Workflow.DataStore.PostgreSQL/PostgresWorkflowDefinitionStore.cs

@@ -0,0 +1,180 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+using StellaOps.Workflow.Abstractions;
+
+namespace StellaOps.Workflow.DataStore.PostgreSQL;
+
+/// <summary>
+/// PostgreSQL-backed implementation of <see cref="IWorkflowDefinitionStore"/>.
+/// Stores versioned canonical workflow definitions in the <c>wf_definitions</c> table.
+/// </summary>
+public sealed class PostgresWorkflowDefinitionStore(
+    PostgresWorkflowDatabase db) : IWorkflowDefinitionStore
+{
+    public async Task<WorkflowDefinitionRecord?> GetActiveAsync(
+        string workflowName, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: false, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $"SELECT * FROM {table} WHERE workflow_name = @name AND is_active = true LIMIT 1");
+        cmd.Parameters.AddWithValue("name", workflowName);
+
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+        return await reader.ReadAsync(cancellationToken) ? ReadRecord(reader) : null;
+    }
+
+    public async Task<WorkflowDefinitionRecord?> GetAsync(
+        string workflowName, string version, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: false, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $"SELECT * FROM {table} WHERE workflow_name = @name AND workflow_version = @ver LIMIT 1");
+        cmd.Parameters.AddWithValue("name", workflowName);
+        cmd.Parameters.AddWithValue("ver", version);
+
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+        return await reader.ReadAsync(cancellationToken) ? ReadRecord(reader) : null;
+    }
+
+    public async Task<IReadOnlyCollection<WorkflowDefinitionRecord>> GetVersionsAsync(
+        string workflowName, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: false, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $"SELECT * FROM {table} WHERE workflow_name = @name ORDER BY created_on_utc DESC");
+        cmd.Parameters.AddWithValue("name", workflowName);
+
+        var results = new List<WorkflowDefinitionRecord>();
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+        while (await reader.ReadAsync(cancellationToken))
+        {
+            results.Add(ReadRecord(reader));
+        }
+        return results;
+    }
+
+    public async Task<IReadOnlyCollection<WorkflowDefinitionRecord>> GetAllActiveAsync(
+        CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: false, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $"SELECT * FROM {table} WHERE is_active = true ORDER BY workflow_name");
+
+        var results = new List<WorkflowDefinitionRecord>();
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+        while (await reader.ReadAsync(cancellationToken))
+        {
+            results.Add(ReadRecord(reader));
+        }
+        return results;
+    }
+
+    public async Task UpsertAsync(
+        WorkflowDefinitionRecord record, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: true, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $@"INSERT INTO {table}
+                (workflow_name, workflow_version, base_version, build_iteration,
+                 content_hash, canonical_definition_json, display_name, is_active,
+                 created_on_utc, activated_on_utc, imported_by)
+               VALUES
+                (@name, @ver, @base_ver, @build_iter,
+                 @hash, @json, @display, @active,
+                 @created, @activated, @imported_by)
+               ON CONFLICT (workflow_name, workflow_version) DO UPDATE SET
+                content_hash = EXCLUDED.content_hash,
+                canonical_definition_json = EXCLUDED.canonical_definition_json,
+                display_name = EXCLUDED.display_name,
+                is_active = EXCLUDED.is_active,
+                activated_on_utc = EXCLUDED.activated_on_utc,
+                imported_by = EXCLUDED.imported_by",
+            scope.Transaction);
+
+        cmd.Parameters.AddWithValue("name", record.WorkflowName);
+        cmd.Parameters.AddWithValue("ver", record.WorkflowVersion);
+        cmd.Parameters.AddWithValue("base_ver", record.BaseVersion);
+        cmd.Parameters.AddWithValue("build_iter", record.BuildIteration);
+        cmd.Parameters.AddWithValue("hash", record.ContentHash);
+        cmd.Parameters.AddWithValue("json", record.CanonicalDefinitionJson);
+        cmd.Parameters.AddWithValue("display", (object?)record.DisplayName ?? DBNull.Value);
+        cmd.Parameters.AddWithValue("active", record.IsActive);
+        cmd.Parameters.AddWithValue("created", record.CreatedOnUtc == default ? DateTime.UtcNow : record.CreatedOnUtc);
+        cmd.Parameters.AddWithValue("activated", (object?)record.ActivatedOnUtc ?? DBNull.Value);
+        cmd.Parameters.AddWithValue("imported_by", (object?)record.ImportedBy ?? DBNull.Value);
+
+        await cmd.ExecuteNonQueryAsync(cancellationToken);
+        await scope.CommitAsync(cancellationToken);
+    }
+
+    public async Task ActivateAsync(
+        string workflowName, string version, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: true, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        // Deactivate all versions
+        await using var deactivate = db.CreateCommand(scope.Connection,
+            $"UPDATE {table} SET is_active = false WHERE workflow_name = @name",
+            scope.Transaction);
+        deactivate.Parameters.AddWithValue("name", workflowName);
+        await deactivate.ExecuteNonQueryAsync(cancellationToken);
+
+        // Activate requested version
+        await using var activate = db.CreateCommand(scope.Connection,
+            $"UPDATE {table} SET is_active = true, activated_on_utc = @now WHERE workflow_name = @name AND workflow_version = @ver",
+            scope.Transaction);
+        activate.Parameters.AddWithValue("name", workflowName);
+        activate.Parameters.AddWithValue("ver", version);
+        activate.Parameters.AddWithValue("now", DateTime.UtcNow);
+        await activate.ExecuteNonQueryAsync(cancellationToken);
+
+        await scope.CommitAsync(cancellationToken);
+    }
+
+    public async Task<WorkflowDefinitionRecord?> FindByHashAsync(
+        string workflowName, string contentHash, CancellationToken cancellationToken = default)
+    {
+        await using var scope = await db.OpenScopeAsync(requireTransaction: false, cancellationToken);
+        var table = db.Qualify("wf_definitions");
+
+        await using var cmd = db.CreateCommand(scope.Connection,
+            $"SELECT * FROM {table} WHERE workflow_name = @name AND content_hash = @hash LIMIT 1");
+        cmd.Parameters.AddWithValue("name", workflowName);
+        cmd.Parameters.AddWithValue("hash", contentHash);
+
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+        return await reader.ReadAsync(cancellationToken) ? ReadRecord(reader) : null;
+    }
+
+    private static WorkflowDefinitionRecord ReadRecord(Npgsql.NpgsqlDataReader reader)
+    {
+        return new WorkflowDefinitionRecord
+        {
+            WorkflowName = reader.GetString(reader.GetOrdinal("workflow_name")),
+            WorkflowVersion = reader.GetString(reader.GetOrdinal("workflow_version")),
+            BaseVersion = reader.GetString(reader.GetOrdinal("base_version")),
+            BuildIteration = reader.GetInt32(reader.GetOrdinal("build_iteration")),
+            ContentHash = reader.GetString(reader.GetOrdinal("content_hash")),
+            CanonicalDefinitionJson = reader.GetString(reader.GetOrdinal("canonical_definition_json")),
+            DisplayName = reader.IsDBNull(reader.GetOrdinal("display_name")) ? null : reader.GetString(reader.GetOrdinal("display_name")),
+            IsActive = reader.GetBoolean(reader.GetOrdinal("is_active")),
+            CreatedOnUtc = reader.GetDateTime(reader.GetOrdinal("created_on_utc")),
+            ActivatedOnUtc = reader.IsDBNull(reader.GetOrdinal("activated_on_utc")) ? null : reader.GetDateTime(reader.GetOrdinal("activated_on_utc")),
+            ImportedBy = reader.IsDBNull(reader.GetOrdinal("imported_by")) ? null : reader.GetString(reader.GetOrdinal("imported_by")),
+        };
+    }
+}