From 530949edcd230d1e697d040b39f52c2643398a62 Mon Sep 17 00:00:00 2001 From: Ninja Date: Sun, 2 Aug 2026 23:56:06 +0100 Subject: [PATCH] feat: let tenants configure executor approval gates via the API Executor nodes default to autonomous, as before. Tenants can now inspect a workflow version's nodes and set, per executor, whether it runs autonomously or waits for approval, without a redeploy. - GET/PUT/DELETE /workflows/{name}/versions/{version}/nodes[/{executorId}] list nodes with their declared, tenant, and effective gates, and write or drop the calling tenant's policy. Bulk writes are all-or-nothing. - IGatePolicyStore gains a tenant dimension (null = host-wide) plus List and Remove; the runner resolves gates for instance.TenantId. Precedence is instance override, tenant policy, host policy, definition, autonomous. - ApprovalGate.Locked marks a declared gate as the author's floor. Tenant config may tighten it; weakening is refused with 409 and re-tightened on read, so a policy stored by another route cannot un-gate an executor. - WorkflowInspector builds a definition once per version against an inspection context to read its nodes; results are cached. Also fixes SqlServerGatePolicyStore.FindAsync returning only the instance-scoped key instead of falling through to the workflow scope, and marks ApprovalGate.Predicate [JsonIgnore] - that store round-trips gates through JsonSerializer, which would have thrown on the delegate. Co-Authored-By: Claude Opus 5 --- README.md | 78 ++ docs/wiki.md | 78 +- .../SqlServerInfrastructureStores.cs | 61 +- src/Abacus.Run/Abstractions/Approvals.cs | 21 + .../Abstractions/WorkflowDefinition.cs | 49 +- src/Abacus.Run/Api/Endpoints.cs | 74 ++ .../Api/GateConfigurationService.cs | 405 ++++++++++ src/Abacus.Run/Api/HostBuilderExtensions.cs | 7 + src/Abacus.Run/Core/GateEvaluator.cs | 15 +- src/Abacus.Run/Core/GatePolicyRules.cs | 105 +++ src/Abacus.Run/Core/Stores.cs | 24 +- src/Abacus.Run/Core/WorkflowInspector.cs | 48 ++ src/Abacus.Run/Core/WorkflowRunner.cs | 3 +- src/Abacus.Run/Persistence/InMemoryStores.cs | 694 +++++++++--------- .../HostFixture.cs | 80 +- .../TenantGateConfigurationTests.cs | 343 +++++++++ .../CheckpointAndStoreTests.cs | 39 +- .../GateConfigurationTests.cs | 571 ++++++++++++++ .../GateEvaluatorTests.cs | 105 ++- 19 files changed, 2441 insertions(+), 359 deletions(-) create mode 100644 src/Abacus.Run/Api/GateConfigurationService.cs create mode 100644 src/Abacus.Run/Core/GatePolicyRules.cs create mode 100644 src/Abacus.Run/Core/WorkflowInspector.cs create mode 100644 tests/Abacus.Run.IntegrationTests/TenantGateConfigurationTests.cs create mode 100644 tests/Abacus.Run.UnitTests/GateConfigurationTests.cs diff --git a/README.md b/README.md index bb5b92b..68dd1f7 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,32 @@ builder.Services `OrderWorkflow` must implement `IWorkflowDefinition` or `IWorkflowDefinition`. Use `WorkflowBuildContext.Node(...)` to attach host executors and declare approval gates. +### Declaring executor gates + +A node attached with no gate block runs autonomously. Pass a gate block to require a human decision, either always or under a predicate: + +```csharp +public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) +{ + ExecutorBinding validate = context.Node(new Validate("validate")); // autonomous + + ExecutorBinding notify = context.Node(new Notify("notify"), gate => gate + .Mode(ExecutionMode.RequireApproval) + .AssignTo("group:ops") + .ExpiresAfter(TimeSpan.FromHours(4))); + + ExecutorBinding settle = context.Node(new Settle("settle"), gate => gate + .When(order => order.Amount > 25_000m) + .Reason("RegulatedSettlement") + .RequireApprovers(2) + .Locked()); // tenants may not weaken this + + // ... +} +``` + +Every gated node is configurable per tenant at run time unless the author calls `.Locked()`. A locked gate is a floor, not a freeze: a tenant may still tighten it. `RawNode(...)` bindings run outside the executor pipeline and cannot be gated at all. + ## API Surface ### Workflow catalog @@ -97,6 +123,58 @@ builder.Services The start endpoint accepts an optional `version` query parameter and supports `Idempotency-Key` and `Prefer: wait=` headers. +### Per-tenant executor configuration + +- `GET /workflows/{name}/versions/{version}/nodes` +- `PUT /workflows/{name}/versions/{version}/nodes` +- `PUT /workflows/{name}/versions/{version}/nodes/{executorId}` +- `DELETE /workflows/{name}/versions/{version}/nodes/{executorId}` + +Tenants configure whether each executor of a workflow runs autonomously or waits for approval, without a redeploy. The tenant comes from `X-Tenant-Id` or the `tenant_id` claim; `version` accepts a concrete version or `latest`. + +`GET` lists the executor nodes the definition declares, each with its declared gate, the calling tenant's override, and the gate that will actually run: + +```json +{ + "workflowName": "order", "workflowVersion": "1.0.0", "tenantId": "acme", + "nodes": [{ + "executorId": "settle", "executorType": "Settle", + "inputType": "OrderContext", "outputType": "OrderResult", + "configurable": true, "locked": false, + "declared": { "mode": "autonomous", "requiredApprovers": 1, "...": "..." }, + "tenantOverride": { "mode": "requireApproval", "assignees": ["group:finance"], "...": "..." }, + "effective": { "mode": "requireApproval", "assignees": ["group:finance"], "...": "..." }, + "effectiveSource": "tenant" + }] +} +``` + +`PUT` writes the calling tenant's policy for one node, or for several at once via `{ "nodes": { "": { ... } } }`. Only `mode` is required; every other field inherits the declared gate, so the body below keeps the author's assignees and expiry: + +```bash +curl -X PUT http://localhost:5000/workflows/order/versions/1.0.0/nodes/settle \ + -H 'X-Tenant-Id: acme' -H 'Content-Type: application/json' \ + -d '{"mode":"requireApproval","requiredApprovers":2}' +``` + +| Field | Notes | +| --- | --- | +| `mode` | `autonomous` or `requireApproval`. `conditional` is code-only — its predicate cannot be expressed in JSON | +| `reason` | Shown on the resulting approval request | +| `assignees`, `escalationAssignees` | Principal or group identifiers | +| `requiredApprovers` | Quorum; at least 1 | +| `expirySeconds` | Approval window; greater than zero | +| `onExpiry` | `deadStop`, `reject`, `autoApprove`, or `escalate` | +| `allowModification`, `requireSegregationOfDuties` | Booleans | + +`DELETE` drops the override and restores the declared gate. A bulk `PUT` is all-or-nothing: if any node is unknown, not configurable, or refused, nothing is persisted. + +At run time the runner resolves gates for the instance's own tenant, in this order: per-instance override, tenant policy, host-wide policy (a policy written with no tenant), then the definition, then autonomous. Responses report which one applied as `effectiveSource`. + +A gate the author declared with `.Locked()` is a floor. Configuration may tighten it; an attempt to weaken it — downgrading `mode`, lowering `requiredApprovers`, enabling `allowModification`, disabling segregation of duties, or setting `onExpiry` to `autoApprove` — is refused with `409` and persists nothing. The runtime re-applies the floor on read as well, so a policy that reached the store by another route still cannot un-gate a locked executor. + +Policies are keyed by workflow **version**, since executor ids and gates change between versions. A tenant's configuration therefore does not carry forward to a new version, and `POST /instances/{id}/rerun` in restart mode creates the new instance at the *current* version — so a restart after a version bump runs under that version's configuration, or under declared defaults if the tenant has none. + ### Instance operations - `GET /instances/{id}` diff --git a/docs/wiki.md b/docs/wiki.md index e58b385..7831af9 100644 --- a/docs/wiki.md +++ b/docs/wiki.md @@ -17,6 +17,7 @@ This page is the repository-level technical wiki. It documents the implementatio - [Retries and failure classification](#retries-and-failure-classification) - [Checkpoints and resumption](#checkpoints-and-resumption) - [Human approval gates](#human-approval-gates) +- [Tenant executor configuration](#tenant-executor-configuration) - [Events, history, and SSE](#events-history-and-sse) - [HTTP API](#http-api) - [Configuration](#configuration) @@ -491,6 +492,48 @@ Decision outcomes are `Approve`, `Reject`, and `ApproveWithModification`. Modifi Expiry actions are `DeadStop`, `Reject`, `AutoApprove`, and `Escalate`. Approval decisions enforce assignees, quorum, and optional segregation of duties. +A gate may also be declared with `.Locked()`, which prevents tenant configuration from weakening it. See [Tenant executor configuration](#tenant-executor-configuration). + +## Tenant executor configuration + +The gate declared in code is a default, not a fixed setting. Each tenant decides whether an executor runs autonomously or waits for approval, through the API and without a redeploy. An executor attached with `context.Node(executor)` and no gate block is autonomous for every tenant until someone changes it. + +### Resolving the effective gate + +The runner resolves each executor's gate for the instance's own tenant, highest precedence first: + +| Precedence | Scope | Written by | +| --- | --- | --- | +| 1 | Per-instance override | `IGatePolicyStore.SetInstanceOverrideAsync` | +| 2 | Tenant policy | `PUT /workflows/{name}/versions/{version}/nodes` with a tenant | +| 3 | Host-wide policy | `IGatePolicyStore.SetAsync` with a null tenant | +| 4 | Workflow definition | The gate block in `BuildAsync` | +| 5 | Host default | `ExecutionMode.Autonomous` | + +A policy-store failure falls back to the definition's gate, never to autonomous: a lookup error must not un-gate a protected executor. `GET` responses report which scope applied as `effectiveSource` (`tenant`, `host`, or `definition`). + +### Locked gates + +`.Locked()` marks a declared gate as the author's floor. Tenant configuration may still tighten it, but any of the following is refused with `409` and persists nothing: + +- Downgrading `Mode` (`RequireApproval` or `Conditional` to `Autonomous`, or `RequireApproval` to `Conditional`). +- Lowering `RequiredApprovers` below the declared quorum. +- Enabling `AllowModification` where the declaration disabled it. +- Disabling `RequireSegregationOfDuties` where the declaration required it. +- Setting `OnExpiry` to `AutoApprove` where the declaration did not. + +The same rules are re-applied when the gate is read at execution time, so a policy that reached the store before the gate was locked — or through a store client rather than the API — still cannot un-gate the executor. A locked `Conditional` gate also keeps its predicate, because a predicate is code and no stored policy can supply one. + +### Scope and lifetime + +Policies are keyed by tenant, workflow name, and workflow **version**, because executor ids and gates change between versions. A tenant's configuration does not carry forward when a new version is registered; instances of the new version run under its declared defaults until configured. `POST /instances/{id}/rerun` in restart mode creates the new instance at the current version, so a restart after a version bump uses that version's configuration. + +Configuration writes are recorded in the audit store as `gate.policy.set` and `gate.policy.reset`, with the actor, tenant, workflow, version, and executor. + +### Discovering nodes + +`GET /workflows/{name}/versions/{version}/nodes` builds the definition once against an inspection context — attaching nothing to a runtime and running no executor — and caches the result per version. Each node reports its executor id, implementation type, input and output types, whether it is configurable, whether it is locked, and its declared, tenant, and effective gates. `RawNode(...)` bindings are listed with `configurable: false`; they run outside the executor middleware pipeline and cannot be approval-gated. + ## Events, history, and SSE Every instance event has a monotonically increasing per-instance `Sequence`. The same sequence is used as the SSE event ID, which lets clients reconnect with `Last-Event-ID` and request replay from the same cursor. @@ -544,6 +587,27 @@ A start request body contains a workflow-specific context object. Optional reque A successful asynchronous start returns `202 Accepted` and a location such as `/instances/{id}`. Unknown workflows return `404`; invalid context returns `400` with field errors. +### Executor nodes and tenant policy + +| Method | Path | Purpose | +| --- | --- | --- | +| `GET` | `/workflows/{name}/versions/{version}/nodes` | List executor nodes with declared, tenant, and effective gates | +| `PUT` | `/workflows/{name}/versions/{version}/nodes` | Configure several nodes in one all-or-nothing write | +| `PUT` | `/workflows/{name}/versions/{version}/nodes/{executorId}` | Configure one node | +| `DELETE` | `/workflows/{name}/versions/{version}/nodes/{executorId}` | Drop the tenant's override for one node | + +The tenant is taken from the `X-Tenant-Id` header or the `tenant_id` claim. `version` accepts a concrete version or `latest`. + +A policy body requires `mode` (`autonomous` or `requireApproval`; `conditional` is code-only) and optionally `reason`, `assignees`, `requiredApprovers`, `expirySeconds`, `onExpiry`, `escalationAssignees`, `allowModification`, and `requireSegregationOfDuties`. Omitted fields inherit the declared gate. + +```bash +curl -X PUT http://localhost:5000/workflows/order/versions/1.0.0/nodes/submit-payment \ + -H 'X-Tenant-Id: acme' -H 'Content-Type: application/json' \ + -d '{"mode":"requireApproval","assignees":["group:finance"],"requiredApprovers":2}' +``` + +Unknown workflow, version, or executor returns `404`. A raw node or an unsupported mode returns `400`. A policy that would weaken a locked gate returns `409`. See [Tenant executor configuration](#tenant-executor-configuration) for precedence and locking rules. + ### Instances and diagnostics | Method | Path | Purpose | @@ -672,7 +736,9 @@ The built-in request/response logging middleware samples bodies and applies an a ### Tenancy and authorization -Instances and approvals carry a tenant identifier. The API and store implementations must enforce tenant isolation at the request boundary and persistence boundary. The in-memory reference stores model the contract but are not a replacement for a production identity and authorization system. +Instances, approvals, and gate policies carry a tenant identifier. The API and store implementations must enforce tenant isolation at the request boundary and persistence boundary. The in-memory reference stores model the contract but are not a replacement for a production identity and authorization system. + +Because gate policies decide whether an executor runs without a human decision, the node configuration endpoints are privileged: authorize them for tenant administrators rather than for anyone who can start an instance. The host derives the tenant from `X-Tenant-Id` when no `tenant_id` claim is present, which is a development convenience — in production, bind the tenant to the authenticated principal so a caller cannot configure another tenant's workflows by setting a header. ### Secrets @@ -862,6 +928,8 @@ Implement `IWorkflowMiddleware` for run-wide behavior or `IExecutorMiddleware` f Provide an `IGatePolicyStore` implementation when approval requirements depend on tenant, workflow, executor, amount, role, or environment. Keep policy evaluation deterministic and observable. +`FindAsync` resolves one executor's gate for a tenant and must itself apply the scope precedence — per-instance override, then the tenant's policy, then the host-wide policy stored under a null tenant — returning `null` when no policy exists so the definition's gate stands. `ListAsync` returns exactly one scope's entries without merging, which is what lets the API show a tenant's own overrides separately from the host default. The locked-gate floor is enforced by the runtime on top of whatever the store returns, so a custom implementation cannot accidentally weaken a protected executor. + ### Custom event sinks and buses The runner publishes through `IEventSink`. A sink may persist the event, relay it to an event bus, or do both. Preserve the per-instance sequence when forwarding to SSE or external consumers. @@ -919,6 +987,14 @@ Check that `AddBackgroundServices()` is registered and that the dispatcher is ru Inspect the approval state, decision authorization, quorum, expiry, and instance status. A successful decision wakes the instance by moving it to a claimable state; the dispatcher must be running to execute the resumed run. +### An executor pauses for approval although the definition left it autonomous + +A tenant or host-wide policy is gating it. Call `GET /workflows/{name}/versions/{version}/nodes` as that tenant and read `effectiveSource`: `tenant` means the tenant configured it, `host` means a host-wide policy applies. `DELETE` the node's override to restore the declared gate. Remember the policy is version-scoped — check the version the instance actually pinned, not the latest. + +### A tenant's configuration appears to be ignored + +Confirm the tenant used to configure the workflow is the tenant the instance runs under; the runner resolves gates for `instance.TenantId`, not for the caller who last edited the policy. Also check the instance's workflow version against the version the policy was written for, and whether a per-instance override outranks it. + ### An outbound call is blocked Check the URL scheme, whether the target resolves to an internal address, and whether the hostname matches `WorkflowHost:Egress:AllowedHosts`. Keep `Egress:Enforce=true` unless this is a controlled local test. diff --git a/src/Abacus.Run.Service/Infrastructure/SqlServerInfrastructureStores.cs b/src/Abacus.Run.Service/Infrastructure/SqlServerInfrastructureStores.cs index 5ee7a53..96a0fdb 100644 --- a/src/Abacus.Run.Service/Infrastructure/SqlServerInfrastructureStores.cs +++ b/src/Abacus.Run.Service/Infrastructure/SqlServerInfrastructureStores.cs @@ -99,20 +99,69 @@ public async ValueTask DeleteAsync(string uri, CancellationToken cancellationTok public sealed class SqlServerGatePolicyStore(IDbContextFactory factory) : IGatePolicyStore { - public async ValueTask FindAsync(string workflowName, string workflowVersion, string executorId, string? instanceId, CancellationToken cancellationToken) + private const string HostScope = "*"; + + public async ValueTask FindAsync(string? tenantId, string workflowName, string workflowVersion, string executorId, string? instanceId, CancellationToken cancellationToken) + { + // Highest scope first, falling through to the next when nothing is stored at that level. + var keys = new List(3); + if (instanceId is { Length: > 0 }) keys.Add($"gate-instance:{instanceId}:{executorId}"); + if (tenantId is { Length: > 0 }) keys.Add(PolicyKey(tenantId, workflowName, workflowVersion, executorId)); + keys.Add(PolicyKey(null, workflowName, workflowVersion, executorId)); + + await using AbacusDbContext db = await factory.CreateDbContextAsync(cancellationToken); + List rows = await db.JsonRows.AsNoTracking() + .Where(item => item.Kind == "gate" && keys.Contains(item.Key)) + .ToListAsync(cancellationToken); + + foreach (string key in keys) + { + JsonRow? row = rows.Find(item => item.Key == key); + if (row is not null) return JsonSerializer.Deserialize(row.Payload); + } + + return null; + } + + public async ValueTask> ListAsync(string? tenantId, string workflowName, string workflowVersion, CancellationToken cancellationToken) { + string prefix = PolicyKey(tenantId, workflowName, workflowVersion, string.Empty); + await using AbacusDbContext db = await factory.CreateDbContextAsync(cancellationToken); - string key = instanceId is null ? $"gate:{workflowName}:{workflowVersion}:{executorId}" : $"gate-instance:{instanceId}:{executorId}"; - JsonRow? row = await db.JsonRows.AsNoTracking().SingleOrDefaultAsync(item => item.Kind == "gate" && item.Key == key, cancellationToken); - return row is null ? null : JsonSerializer.Deserialize(row.Payload); + List rows = await db.JsonRows.AsNoTracking() + .Where(item => item.Kind == "gate" && item.Key.StartsWith(prefix)) + .ToListAsync(cancellationToken); + + var gates = new Dictionary(StringComparer.Ordinal); + foreach (JsonRow row in rows) + { + ApprovalGate? gate = JsonSerializer.Deserialize(row.Payload); + if (gate is not null) gates[row.Key[prefix.Length..]] = gate; + } + + return gates; } - public ValueTask SetAsync(string workflowName, string workflowVersion, string executorId, ApprovalGate gate, CancellationToken cancellationToken) - => SetCoreAsync($"gate:{workflowName}:{workflowVersion}:{executorId}", gate, cancellationToken); + public ValueTask SetAsync(string? tenantId, string workflowName, string workflowVersion, string executorId, ApprovalGate gate, CancellationToken cancellationToken) + => SetCoreAsync(PolicyKey(tenantId, workflowName, workflowVersion, executorId), gate, cancellationToken); + + public async ValueTask RemoveAsync(string? tenantId, string workflowName, string workflowVersion, string executorId, CancellationToken cancellationToken) + { + string key = PolicyKey(tenantId, workflowName, workflowVersion, executorId); + await using AbacusDbContext db = await factory.CreateDbContextAsync(cancellationToken); + JsonRow? row = await db.JsonRows.SingleOrDefaultAsync(item => item.Kind == "gate" && item.Key == key, cancellationToken); + if (row is null) return false; + db.JsonRows.Remove(row); + await db.SaveChangesAsync(cancellationToken); + return true; + } public ValueTask SetInstanceOverrideAsync(string instanceId, string executorId, ApprovalGate gate, CancellationToken cancellationToken) => SetCoreAsync($"gate-instance:{instanceId}:{executorId}", gate, cancellationToken); + private static string PolicyKey(string? tenantId, string workflowName, string workflowVersion, string executorId) + => $"gate:{(string.IsNullOrEmpty(tenantId) ? HostScope : tenantId)}:{workflowName}:{workflowVersion}:{executorId}"; + private async ValueTask SetCoreAsync(string key, ApprovalGate gate, CancellationToken cancellationToken) { await using AbacusDbContext db = await factory.CreateDbContextAsync(cancellationToken); diff --git a/src/Abacus.Run/Abstractions/Approvals.cs b/src/Abacus.Run/Abstractions/Approvals.cs index 874e9cd..f62d29a 100644 --- a/src/Abacus.Run/Abstractions/Approvals.cs +++ b/src/Abacus.Run/Abstractions/Approvals.cs @@ -1,3 +1,5 @@ +using System.Text.Json.Serialization; + namespace Abacus.Run.Abstractions; /// Whether an executor runs on its own or pauses for a human decision (PRD FR-9.1). @@ -43,6 +45,11 @@ public sealed record ApprovalGate public ExecutionMode Mode { get; init; } = ExecutionMode.Autonomous; /// Evaluated only when is . + /// + /// Never serialized: gates round-trip through the policy store as data, and a predicate only + /// exists on the gate the definition itself declared. + /// + [JsonIgnore] public Func>? Predicate { get; init; } public string? Reason { get; init; } @@ -53,6 +60,13 @@ public sealed record ApprovalGate public IReadOnlyList EscalationAssignees { get; init; } = []; public bool AllowModification { get; init; } public bool RequireSegregationOfDuties { get; init; } + + /// + /// Marks this gate as the author's floor. Tenant configuration may still tighten a locked gate, + /// but may never loosen it — the API rejects such a write and the evaluator re-tightens anything + /// that reached the policy store by another route. + /// + public bool Locked { get; init; } } public sealed class ApprovalGateBuilder @@ -128,6 +142,13 @@ public ApprovalGateBuilder RequireSegregationOfDuties(bool require = true) return this; } + /// Forbids tenant configuration from loosening this gate (PRD FR-9.2). + public ApprovalGateBuilder Locked(bool locked = true) + { + _gate = _gate with { Locked = locked }; + return this; + } + public ApprovalGate Build() => _gate; } diff --git a/src/Abacus.Run/Abstractions/WorkflowDefinition.cs b/src/Abacus.Run/Abstractions/WorkflowDefinition.cs index 938bec9..35052c5 100644 --- a/src/Abacus.Run/Abstractions/WorkflowDefinition.cs +++ b/src/Abacus.Run/Abstractions/WorkflowDefinition.cs @@ -27,6 +27,22 @@ FailureDisposition IWorkflowDefinition.Classify(WorkflowFailure failure) => DefaultFailureClassifier.Instance.Classify(failure); } +/// +/// One executor node of a workflow graph as the definition declared it. Produced during a build and +/// surfaced by the catalog API so a tenant can see what there is to configure. +/// +/// +/// Configurable is false for bindings: they run +/// outside the host executor pipeline and therefore cannot carry an approval gate at all. +/// +public sealed record WorkflowNodeDescriptor( + string ExecutorId, + string ExecutorType, + string? InputType, + string? OutputType, + ApprovalGate DeclaredGate, + bool Configurable); + /// /// Supplied to . The single place approval gates are /// declared and the only supported way to attach an executor to the host runtime. @@ -63,7 +79,14 @@ public WorkflowBuildContext( /// Gates declared during this build, by executor id. Read by the runtime. public IReadOnlyDictionary Gates => _gates; + /// + /// Every node attached during this build, in declaration order. Read by the catalog API to + /// describe what a tenant may configure. + /// + public IReadOnlyList Nodes => _nodes; + private readonly Dictionary _gates = []; + private readonly List _nodes = []; /// /// Attaches a host executor: wires the middleware pipeline, the gate, and the instance runtime, @@ -82,6 +105,10 @@ public ExecutorBinding Node(IHostExecutor executor, Action? } _gates[executor.Id] = configured; + Record(new WorkflowNodeDescriptor( + executor.Id, executor.GetType().Name, executor.InputType.Name, executor.OutputType.Name, + configured, Configurable: true)); + return _attach(executor, configured); } @@ -100,12 +127,30 @@ public ExecutorBinding RawNode(ExecutorBinding binding, Action; convert the executor or remove the gate."); } + Record(new WorkflowNodeDescriptor( + binding.Id, binding.GetType().Name, null, null, ApprovalGate.Autonomous, Configurable: false)); + return binding; } + /// Last declaration of an id wins, matching how is built. + private void Record(WorkflowNodeDescriptor node) + { + int existing = _nodes.FindIndex(n => string.Equals(n.ExecutorId, node.ExecutorId, StringComparison.Ordinal)); + if (existing >= 0) + { + _nodes[existing] = node; + } + else + { + _nodes.Add(node); + } + } + /// Build context for read-only inspection (graph rendering), with no runtime attachment. - public static WorkflowBuildContext ForInspection(string workflowName, string workflowVersion) - => new("inspection", "inspection", workflowName, workflowVersion, 0, null, + public static WorkflowBuildContext ForInspection( + string workflowName, string workflowVersion, IServiceProvider? services = null) + => new("inspection", "inspection", workflowName, workflowVersion, 0, services, (executor, _) => { executor.Runtime = HostExecutorRuntime.Unattached; diff --git a/src/Abacus.Run/Api/Endpoints.cs b/src/Abacus.Run/Api/Endpoints.cs index 3b3103b..e221845 100644 --- a/src/Abacus.Run/Api/Endpoints.cs +++ b/src/Abacus.Run/Api/Endpoints.cs @@ -107,6 +107,65 @@ private static void MapCatalog(IEndpointRouteBuilder app) }) }); }); + + MapNodeConfiguration(app); + } + + /// + /// The executor nodes of one workflow version, and the tenant's execution policy over them. + /// Nodes are autonomous unless the definition declared a gate or the tenant configured one. + /// + private static void MapNodeConfiguration(IEndpointRouteBuilder app) + { + app.MapGet("/workflows/{name}/versions/{version}/nodes", async ( + string name, string version, HttpContext http, IGateConfigurationService config, CancellationToken cancellationToken) => + (await config.GetNodesAsync(name, version, http.TenantId(), cancellationToken).ConfigureAwait(false)) + .ToHttpResult()); + + app.MapPut("/workflows/{name}/versions/{version}/nodes", async ( + string name, string version, NodeConfigurationRequestDto? body, HttpContext http, + IGateConfigurationService config, CancellationToken cancellationToken) => + { + if (body?.Nodes is null) + { + return Results.ValidationProblem(new Dictionary + { + ["nodes"] = ["A map of executor id to policy is required."] + }); + } + + return (await config + .SetNodesAsync(name, version, http.TenantId(), body.Nodes, http.User, cancellationToken) + .ConfigureAwait(false)) + .ToHttpResult(); + }); + + app.MapPut("/workflows/{name}/versions/{version}/nodes/{executorId}", async ( + string name, string version, string executorId, ExecutionPolicyDto? body, HttpContext http, + IGateConfigurationService config, CancellationToken cancellationToken) => + { + if (body is null) + { + return Results.ValidationProblem(new Dictionary + { + ["mode"] = ["A policy body is required."] + }); + } + + return (await config + .SetNodesAsync(name, version, http.TenantId(), + new Dictionary { [executorId] = body }, http.User, cancellationToken) + .ConfigureAwait(false)) + .ToHttpResult(); + }); + + app.MapDelete("/workflows/{name}/versions/{version}/nodes/{executorId}", async ( + string name, string version, string executorId, HttpContext http, + IGateConfigurationService config, CancellationToken cancellationToken) => + (await config + .ResetNodeAsync(name, version, http.TenantId(), executorId, http.User, cancellationToken) + .ConfigureAwait(false)) + .ToHttpResult()); } private static void MapInstances(IEndpointRouteBuilder app) @@ -435,6 +494,21 @@ internal static bool TryParseOutcome(string? value, out ApprovalOutcomeKind outc approval.Assignees, approval.RequiredApprovers, approval.AllowModification, approval.CreatedAt, approval.ExpiresAt, $"/approvals/{approval.ApprovalId}/decision"); + public static IResult ToHttpResult(this GateConfigResult result) => result.Kind switch + { + GateConfigResultKind.Ok => Results.Ok(result.Nodes), + GateConfigResultKind.UnknownWorkflow => Results.NotFound(), + GateConfigResultKind.UnknownExecutor => Results.Problem( + title: "Unknown executor", detail: result.Detail, statusCode: StatusCodes.Status404NotFound), + GateConfigResultKind.NotConfigurable => Results.Problem( + title: "Executor is not configurable", detail: result.Detail, statusCode: StatusCodes.Status400BadRequest), + GateConfigResultKind.Rejected => Results.Problem( + title: "Policy rejected by the workflow definition", detail: result.Detail, + statusCode: StatusCodes.Status409Conflict), + _ => Results.ValidationProblem( + result.Errors?.ToDictionary(kv => kv.Key, kv => kv.Value) ?? new Dictionary()) + }; + public static IResult ToHttpResult(this ControlResult result) => result.Kind switch { ControlResultKind.Accepted => Results.Accepted( diff --git a/src/Abacus.Run/Api/GateConfigurationService.cs b/src/Abacus.Run/Api/GateConfigurationService.cs new file mode 100644 index 0000000..0902be2 --- /dev/null +++ b/src/Abacus.Run/Api/GateConfigurationService.cs @@ -0,0 +1,405 @@ +using System.Security.Claims; +using System.Text.Json; +using Abacus.Run.Abstractions; +using Abacus.Run.Core; + +namespace Abacus.Run.Api; + +/// +/// A tenant-supplied execution policy for one executor. Every field except Mode is optional +/// and falls back to the value the workflow definition declared, so +/// {"mode":"requireApproval"} keeps the author's assignees, quorum and expiry. +/// +public sealed record ExecutionPolicyDto( + string Mode, + string? Reason = null, + IReadOnlyList? Assignees = null, + int? RequiredApprovers = null, + int? ExpirySeconds = null, + string? OnExpiry = null, + IReadOnlyList? EscalationAssignees = null, + bool? AllowModification = null, + bool? RequireSegregationOfDuties = null); + +/// Bulk configuration body: executor id to policy. +public sealed record NodeConfigurationRequestDto(IReadOnlyDictionary Nodes); + +/// Where the gate an instance would actually run under came from. +public static class GateSources +{ + public const string Definition = "definition"; + public const string Host = "host"; + public const string Tenant = "tenant"; +} + +public sealed record ExecutorNodeDto( + string ExecutorId, + string ExecutorType, + string? InputType, + string? OutputType, + bool Configurable, + bool Locked, + ExecutionPolicyDto Declared, + ExecutionPolicyDto? TenantOverride, + ExecutionPolicyDto Effective, + string EffectiveSource); + +public sealed record WorkflowNodesDto( + string WorkflowName, + string WorkflowVersion, + string TenantId, + IReadOnlyList Nodes); + +public enum GateConfigResultKind +{ + Ok, + UnknownWorkflow, + UnknownExecutor, + NotConfigurable, + Invalid, + Rejected +} + +public sealed record GateConfigResult( + GateConfigResultKind Kind, + WorkflowNodesDto? Nodes = null, + IReadOnlyDictionary? Errors = null, + string? Detail = null); + +public interface IGateConfigurationService +{ + Task GetNodesAsync( + string workflowName, string? version, string tenantId, CancellationToken cancellationToken); + + Task SetNodesAsync( + string workflowName, string? version, string tenantId, + IReadOnlyDictionary policies, ClaimsPrincipal? user, CancellationToken cancellationToken); + + Task ResetNodeAsync( + string workflowName, string? version, string tenantId, string executorId, + ClaimsPrincipal? user, CancellationToken cancellationToken); +} + +/// +/// Reads a workflow's declared executor nodes and applies a tenant's execution policy over them. +/// +/// +/// Writes are all-or-nothing: every policy in a request is validated against the declaration before +/// any of them is persisted, so a rejected node cannot leave a half-applied configuration behind. +/// +public sealed class GateConfigurationService : IGateConfigurationService +{ + private readonly IWorkflowRegistry _registry; + private readonly IWorkflowInspector _inspector; + private readonly IGatePolicyStore _policies; + private readonly IAuditStore? _audit; + private readonly TimeProvider _clock; + + public GateConfigurationService( + IWorkflowRegistry registry, + IWorkflowInspector inspector, + IGatePolicyStore policies, + IAuditStore? audit = null, + TimeProvider? clock = null) + { + _registry = registry ?? throw new ArgumentNullException(nameof(registry)); + _inspector = inspector ?? throw new ArgumentNullException(nameof(inspector)); + _policies = policies ?? throw new ArgumentNullException(nameof(policies)); + _audit = audit; + _clock = clock ?? TimeProvider.System; + } + + public async Task GetNodesAsync( + string workflowName, string? version, string tenantId, CancellationToken cancellationToken) + { + WorkflowDescriptor? descriptor = Resolve(workflowName, version); + return descriptor is null + ? new GateConfigResult(GateConfigResultKind.UnknownWorkflow) + : new GateConfigResult(GateConfigResultKind.Ok, + await ProjectAsync(descriptor, tenantId, cancellationToken).ConfigureAwait(false)); + } + + public async Task SetNodesAsync( + string workflowName, string? version, string tenantId, + IReadOnlyDictionary policies, ClaimsPrincipal? user, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(policies); + + WorkflowDescriptor? descriptor = Resolve(workflowName, version); + if (descriptor is null) + { + return new GateConfigResult(GateConfigResultKind.UnknownWorkflow); + } + + if (policies.Count == 0) + { + return new GateConfigResult(GateConfigResultKind.Invalid, Errors: new Dictionary + { + ["nodes"] = ["At least one executor policy is required."] + }); + } + + IReadOnlyList nodes = + await _inspector.InspectAsync(descriptor, cancellationToken).ConfigureAwait(false); + + var errors = new Dictionary(StringComparer.Ordinal); + var resolved = new List<(WorkflowNodeDescriptor Node, ApprovalGate Gate)>(policies.Count); + + foreach ((string executorId, ExecutionPolicyDto policy) in policies) + { + WorkflowNodeDescriptor? node = nodes + .FirstOrDefault(n => string.Equals(n.ExecutorId, executorId, StringComparison.Ordinal)); + + if (node is null) + { + return new GateConfigResult(GateConfigResultKind.UnknownExecutor, + Detail: $"Workflow '{descriptor.Name}' version '{descriptor.Version}' has no executor '{executorId}'."); + } + + if (!node.Configurable) + { + return new GateConfigResult(GateConfigResultKind.NotConfigurable, + Detail: $"Executor '{executorId}' is bound as a raw node and cannot be approval-gated."); + } + + if (!TryBuildGate(node.DeclaredGate, policy, out ApprovalGate gate, out string[] policyErrors)) + { + errors[executorId] = policyErrors; + continue; + } + + IReadOnlyList violations = GatePolicyRules.Violations(node.DeclaredGate, gate); + if (violations.Count > 0) + { + return new GateConfigResult(GateConfigResultKind.Rejected, + Detail: $"Executor '{executorId}' is locked by the workflow definition. " + string.Join(" ", violations)); + } + + resolved.Add((node, gate)); + } + + if (errors.Count > 0) + { + return new GateConfigResult(GateConfigResultKind.Invalid, Errors: errors); + } + + foreach ((WorkflowNodeDescriptor node, ApprovalGate gate) in resolved) + { + await _policies + .SetAsync(tenantId, descriptor.Name, descriptor.Version, node.ExecutorId, gate, cancellationToken) + .ConfigureAwait(false); + + await AuditAsync("gate.policy.set", descriptor, tenantId, node.ExecutorId, user, gate.Mode.ToString(), cancellationToken) + .ConfigureAwait(false); + } + + return new GateConfigResult(GateConfigResultKind.Ok, + await ProjectAsync(descriptor, tenantId, cancellationToken).ConfigureAwait(false)); + } + + public async Task ResetNodeAsync( + string workflowName, string? version, string tenantId, string executorId, + ClaimsPrincipal? user, CancellationToken cancellationToken) + { + WorkflowDescriptor? descriptor = Resolve(workflowName, version); + if (descriptor is null) + { + return new GateConfigResult(GateConfigResultKind.UnknownWorkflow); + } + + IReadOnlyList nodes = + await _inspector.InspectAsync(descriptor, cancellationToken).ConfigureAwait(false); + + if (!nodes.Any(n => string.Equals(n.ExecutorId, executorId, StringComparison.Ordinal))) + { + return new GateConfigResult(GateConfigResultKind.UnknownExecutor, + Detail: $"Workflow '{descriptor.Name}' version '{descriptor.Version}' has no executor '{executorId}'."); + } + + await _policies + .RemoveAsync(tenantId, descriptor.Name, descriptor.Version, executorId, cancellationToken) + .ConfigureAwait(false); + + await AuditAsync("gate.policy.reset", descriptor, tenantId, executorId, user, null, cancellationToken) + .ConfigureAwait(false); + + return new GateConfigResult(GateConfigResultKind.Ok, + await ProjectAsync(descriptor, tenantId, cancellationToken).ConfigureAwait(false)); + } + + private WorkflowDescriptor? Resolve(string workflowName, string? version) + => _registry.Resolve( + workflowName, + string.IsNullOrWhiteSpace(version) || string.Equals(version, "latest", StringComparison.OrdinalIgnoreCase) + ? null + : version); + + private async Task ProjectAsync( + WorkflowDescriptor descriptor, string tenantId, CancellationToken cancellationToken) + { + IReadOnlyList nodes = + await _inspector.InspectAsync(descriptor, cancellationToken).ConfigureAwait(false); + + IReadOnlyDictionary tenantPolicies = await _policies + .ListAsync(tenantId, descriptor.Name, descriptor.Version, cancellationToken).ConfigureAwait(false); + + IReadOnlyDictionary hostPolicies = await _policies + .ListAsync(null, descriptor.Name, descriptor.Version, cancellationToken).ConfigureAwait(false); + + var projected = new List(nodes.Count); + + foreach (WorkflowNodeDescriptor node in nodes) + { + tenantPolicies.TryGetValue(node.ExecutorId, out ApprovalGate? tenantGate); + hostPolicies.TryGetValue(node.ExecutorId, out ApprovalGate? hostGate); + + ApprovalGate? applied = tenantGate ?? hostGate; + string source = tenantGate is not null + ? GateSources.Tenant + : hostGate is not null ? GateSources.Host : GateSources.Definition; + + ApprovalGate effective = applied is null + ? node.DeclaredGate + : GatePolicyRules.Reconcile(node.DeclaredGate, applied); + + projected.Add(new ExecutorNodeDto( + node.ExecutorId, + node.ExecutorType, + node.InputType, + node.OutputType, + node.Configurable, + node.DeclaredGate.Locked, + ToDto(node.DeclaredGate), + tenantGate is null ? null : ToDto(tenantGate), + ToDto(effective), + source)); + } + + return new WorkflowNodesDto(descriptor.Name, descriptor.Version, tenantId, projected); + } + + /// + /// Folds a tenant policy onto the declaration. Conditional is deliberately not accepted: its + /// predicate is code, and no JSON body can supply one. + /// + private static bool TryBuildGate( + ApprovalGate declared, ExecutionPolicyDto? policy, out ApprovalGate gate, out string[] errors) + { + gate = declared; + + if (policy is null) + { + errors = ["A policy body is required."]; + return false; + } + + var problems = new List(); + + if (!TryParseMode(policy.Mode, out ExecutionMode mode)) + { + problems.Add("mode must be one of: autonomous, requireApproval."); + } + + ExpiryAction onExpiry = declared.OnExpiry; + if (policy.OnExpiry is { Length: > 0 } && + !Enum.TryParse(policy.OnExpiry, ignoreCase: true, out onExpiry)) + { + problems.Add("onExpiry must be one of: deadStop, reject, autoApprove, escalate."); + } + + if (policy.RequiredApprovers is { } approvers && approvers < 1) + { + problems.Add("requiredApprovers must be at least 1."); + } + + if (policy.ExpirySeconds is { } seconds && seconds <= 0) + { + problems.Add("expirySeconds must be greater than zero."); + } + + if (problems.Count > 0) + { + errors = [.. problems]; + return false; + } + + gate = declared with + { + Mode = mode, + + // A tenant switching a Conditional gate to always-on (or off) discards the predicate: + // keeping it would silently re-gate invocations the tenant asked to let through. + Predicate = null, + Reason = policy.Reason ?? declared.Reason, + Assignees = policy.Assignees ?? declared.Assignees, + RequiredApprovers = policy.RequiredApprovers ?? declared.RequiredApprovers, + Expiry = policy.ExpirySeconds is { } expiry ? TimeSpan.FromSeconds(expiry) : declared.Expiry, + OnExpiry = onExpiry, + EscalationAssignees = policy.EscalationAssignees ?? declared.EscalationAssignees, + AllowModification = policy.AllowModification ?? declared.AllowModification, + RequireSegregationOfDuties = policy.RequireSegregationOfDuties ?? declared.RequireSegregationOfDuties + }; + + errors = []; + return true; + } + + internal static bool TryParseMode(string? value, out ExecutionMode mode) + { + switch (value?.Trim().ToLowerInvariant()) + { + case "autonomous": + mode = ExecutionMode.Autonomous; + return true; + case "requireapproval": + case "require_approval": + mode = ExecutionMode.RequireApproval; + return true; + default: + mode = default; + return false; + } + } + + private static ExecutionPolicyDto ToDto(ApprovalGate gate) => new( + Mode: gate.Mode switch + { + ExecutionMode.RequireApproval => "requireApproval", + ExecutionMode.Conditional => "conditional", + _ => "autonomous" + }, + Reason: gate.Reason, + Assignees: gate.Assignees, + RequiredApprovers: gate.RequiredApprovers, + ExpirySeconds: (int)gate.Expiry.TotalSeconds, + OnExpiry: gate.OnExpiry.ToString(), + EscalationAssignees: gate.EscalationAssignees, + AllowModification: gate.AllowModification, + RequireSegregationOfDuties: gate.RequireSegregationOfDuties); + + private ValueTask AuditAsync( + string action, WorkflowDescriptor descriptor, string tenantId, string executorId, + ClaimsPrincipal? user, string? mode, CancellationToken cancellationToken) + => _audit is null + ? ValueTask.CompletedTask + : _audit.WriteAsync(new AuditEntry + { + Action = action, + ActorId = ActorOf(user), + Detail = JsonSerializer.Serialize(new + { + tenantId, + workflow = descriptor.Name, + version = descriptor.Version, + executorId, + mode + }), + OccurredAt = _clock.GetUtcNow() + }, cancellationToken); + + private static string ActorOf(ClaimsPrincipal? user) + => user?.FindFirst(ClaimTypes.NameIdentifier)?.Value + ?? user?.FindFirst("sub")?.Value + ?? user?.Identity?.Name + ?? "anonymous"; +} diff --git a/src/Abacus.Run/Api/HostBuilderExtensions.cs b/src/Abacus.Run/Api/HostBuilderExtensions.cs index f374c5e..f7b4b39 100644 --- a/src/Abacus.Run/Api/HostBuilderExtensions.cs +++ b/src/Abacus.Run/Api/HostBuilderExtensions.cs @@ -135,6 +135,13 @@ public static WorkflowHostBuilder AddWorkflowHost( // Runtime services.TryAddSingleton(sp => new WorkflowRegistry(sp.GetServices())); + services.TryAddSingleton(sp => new WorkflowInspector(sp)); + services.TryAddSingleton(sp => new GateConfigurationService( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetService(), + sp.GetRequiredService())); services.TryAddSingleton(sp => new MiddlewarePipelineFactory( sp.GetServices(), sp.GetServices())); services.TryAddSingleton(); diff --git a/src/Abacus.Run/Core/GateEvaluator.cs b/src/Abacus.Run/Core/GateEvaluator.cs index b355577..b0ded3a 100644 --- a/src/Abacus.Run/Core/GateEvaluator.cs +++ b/src/Abacus.Run/Core/GateEvaluator.cs @@ -8,9 +8,11 @@ namespace Abacus.Run.Core; /// or fails. /// /// -/// Precedence (highest first): per-instance override, runtime policy store, workflow definition, -/// host default (). A policy-store failure falls back to the -/// definition's gate — never to Autonomous, which would silently un-gate a protected executor. +/// Precedence (highest first): per-instance override, the running tenant's policy, the host-wide +/// policy, the workflow definition, host default (). A +/// policy-store failure falls back to the definition's gate — never to Autonomous, which would +/// silently un-gate a protected executor. A policy that would loosen a gate the definition marked +/// is re-tightened by . /// public sealed class GateEvaluator : IGateEvaluator { @@ -19,16 +21,19 @@ public sealed class GateEvaluator : IGateEvaluator private readonly IApprovalStore? _approvalStore; private readonly string _workflowName; private readonly string _workflowVersion; + private readonly string? _tenantId; public GateEvaluator( string workflowName, string workflowVersion, + string? tenantId, IReadOnlyDictionary definitionGates, IGatePolicyStore? policyStore = null, IApprovalStore? approvalStore = null) { _workflowName = workflowName; _workflowVersion = workflowVersion; + _tenantId = tenantId; _definitionGates = definitionGates ?? new Dictionary(); _policyStore = policyStore; _approvalStore = approvalStore; @@ -142,10 +147,10 @@ private async ValueTask ResolveGateAsync( try { ApprovalGate? policy = await _policyStore - .FindAsync(_workflowName, _workflowVersion, executorId, instanceId, cancellationToken) + .FindAsync(_tenantId, _workflowName, _workflowVersion, executorId, instanceId, cancellationToken) .ConfigureAwait(false); - return policy ?? definitionGate; + return policy is null ? definitionGate : GatePolicyRules.Reconcile(definitionGate, policy); } catch { diff --git a/src/Abacus.Run/Core/GatePolicyRules.cs b/src/Abacus.Run/Core/GatePolicyRules.cs new file mode 100644 index 0000000..abf5306 --- /dev/null +++ b/src/Abacus.Run/Core/GatePolicyRules.cs @@ -0,0 +1,105 @@ +using Abacus.Run.Abstractions; + +namespace Abacus.Run.Core; + +/// +/// Decides what a tenant-supplied gate may do to the gate the workflow author declared in code. +/// +/// +/// An unlocked declaration is advisory: tenant configuration replaces it outright, in either +/// direction. A declaration marked is the author's floor — +/// configuration may still tighten it, never loosen it. is what the API +/// reports on a rejected write; is the runtime's belt-and-braces equivalent, +/// so a policy written before a gate was locked (or through some other store client) still cannot +/// weaken it at execution time. +/// +public static class GatePolicyRules +{ + /// Higher means more human oversight. + private static int Enforcement(ExecutionMode mode) => mode switch + { + ExecutionMode.RequireApproval => 2, + ExecutionMode.Conditional => 1, + _ => 0 + }; + + /// + /// The ways weakens , empty when it does + /// not. Always empty for an unlocked declaration. + /// + public static IReadOnlyList Violations(ApprovalGate declared, ApprovalGate candidate) + { + ArgumentNullException.ThrowIfNull(declared); + ArgumentNullException.ThrowIfNull(candidate); + + if (!declared.Locked) + { + return []; + } + + var violations = new List(); + + if (Enforcement(candidate.Mode) < Enforcement(declared.Mode)) + { + violations.Add($"Mode cannot be weakened from '{declared.Mode}' to '{candidate.Mode}'."); + } + + if (candidate.RequiredApprovers < declared.RequiredApprovers) + { + violations.Add( + $"RequiredApprovers cannot be lowered below {declared.RequiredApprovers}."); + } + + if (candidate.AllowModification && !declared.AllowModification) + { + violations.Add("AllowModification cannot be enabled."); + } + + if (declared.RequireSegregationOfDuties && !candidate.RequireSegregationOfDuties) + { + violations.Add("RequireSegregationOfDuties cannot be disabled."); + } + + if (candidate.OnExpiry == ExpiryAction.AutoApprove && declared.OnExpiry != ExpiryAction.AutoApprove) + { + violations.Add("OnExpiry cannot be set to AutoApprove."); + } + + return violations; + } + + /// + /// The gate that actually runs: with every field that weakens a + /// locked pulled back to the declaration. + /// + public static ApprovalGate Reconcile(ApprovalGate declared, ApprovalGate candidate) + { + ArgumentNullException.ThrowIfNull(declared); + ArgumentNullException.ThrowIfNull(candidate); + + if (!declared.Locked) + { + return candidate; + } + + bool keepDeclaredMode = Enforcement(candidate.Mode) < Enforcement(declared.Mode); + + return candidate with + { + Mode = keepDeclaredMode ? declared.Mode : candidate.Mode, + + // The predicate only lives on the declaration, so a Conditional floor keeps its own. + Predicate = keepDeclaredMode || candidate.Mode == ExecutionMode.Conditional + ? declared.Predicate + : candidate.Predicate, + + RequiredApprovers = Math.Max(candidate.RequiredApprovers, declared.RequiredApprovers), + AllowModification = candidate.AllowModification && declared.AllowModification, + RequireSegregationOfDuties = candidate.RequireSegregationOfDuties || declared.RequireSegregationOfDuties, + OnExpiry = candidate.OnExpiry == ExpiryAction.AutoApprove && declared.OnExpiry != ExpiryAction.AutoApprove + ? declared.OnExpiry + : candidate.OnExpiry, + Locked = true + }; + } +} diff --git a/src/Abacus.Run/Core/Stores.cs b/src/Abacus.Run/Core/Stores.cs index a07572e..5b1c38e 100644 --- a/src/Abacus.Run/Core/Stores.cs +++ b/src/Abacus.Run/Core/Stores.cs @@ -112,14 +112,32 @@ ValueTask TryRecordDecisionAsync( ValueTask CancelForInstanceAsync(string instanceId, CancellationToken cancellationToken); } -/// Runtime-editable gate policy (PRD FR-9.2), keyed by workflow/version/executor. +/// +/// Runtime-editable gate policy (PRD FR-9.2), keyed by tenant/workflow/version/executor. +/// +/// +/// A null tenantId addresses the host-wide policy, which applies to every tenant that has no +/// policy of its own. Resolution order in is: per-instance override, the +/// tenant's own policy, the host-wide policy, then nothing — leaving the definition's gate to stand. +/// public interface IGatePolicyStore { ValueTask FindAsync( - string workflowName, string workflowVersion, string executorId, string? instanceId, CancellationToken cancellationToken); + string? tenantId, string workflowName, string workflowVersion, string executorId, string? instanceId, + CancellationToken cancellationToken); + + /// Policies written at exactly this scope, by executor id. Never merged with another scope. + ValueTask> ListAsync( + string? tenantId, string workflowName, string workflowVersion, CancellationToken cancellationToken); ValueTask SetAsync( - string workflowName, string workflowVersion, string executorId, ApprovalGate gate, CancellationToken cancellationToken); + string? tenantId, string workflowName, string workflowVersion, string executorId, ApprovalGate gate, + CancellationToken cancellationToken); + + /// Drops the policy at this scope. Returns false when there was nothing to drop. + ValueTask RemoveAsync( + string? tenantId, string workflowName, string workflowVersion, string executorId, + CancellationToken cancellationToken); ValueTask SetInstanceOverrideAsync(string instanceId, string executorId, ApprovalGate gate, CancellationToken cancellationToken); } diff --git a/src/Abacus.Run/Core/WorkflowInspector.cs b/src/Abacus.Run/Core/WorkflowInspector.cs new file mode 100644 index 0000000..7dda842 --- /dev/null +++ b/src/Abacus.Run/Core/WorkflowInspector.cs @@ -0,0 +1,48 @@ +using System.Collections.Concurrent; +using Abacus.Run.Abstractions; + +namespace Abacus.Run.Core; + +/// Describes the executor nodes a registered workflow version declares. +public interface IWorkflowInspector +{ + ValueTask> InspectAsync( + WorkflowDescriptor descriptor, CancellationToken cancellationToken); +} + +/// +/// Builds a definition once against an inspection context to read its nodes back. +/// +/// +/// The graph of a registered version is fixed — the registry rejects two definitions for the same +/// version — so the result is cached per name/version. Building attaches nothing to a runtime and +/// runs no executor: hands every node +/// . +/// +public sealed class WorkflowInspector : IWorkflowInspector +{ + private readonly IServiceProvider? _services; + private readonly ConcurrentDictionary> _cache = + new(StringComparer.OrdinalIgnoreCase); + + public WorkflowInspector(IServiceProvider? services = null) => _services = services; + + public async ValueTask> InspectAsync( + WorkflowDescriptor descriptor, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(descriptor); + + string key = $"{descriptor.Name}|{descriptor.Version}"; + if (_cache.TryGetValue(key, out IReadOnlyList? cached)) + { + return cached; + } + + var context = WorkflowBuildContext.ForInspection(descriptor.Name, descriptor.Version, _services); + await descriptor.Definition.BuildAsync(context, cancellationToken).ConfigureAwait(false); + + IReadOnlyList nodes = [.. context.Nodes]; + _cache[key] = nodes; + return nodes; + } +} diff --git a/src/Abacus.Run/Core/WorkflowRunner.cs b/src/Abacus.Run/Core/WorkflowRunner.cs index 60ef024..5b34a30 100644 --- a/src/Abacus.Run/Core/WorkflowRunner.cs +++ b/src/Abacus.Run/Core/WorkflowRunner.cs @@ -262,7 +262,8 @@ private ExecutorBinding Attach( Attempt = invocation.Attempt, Pipeline = pipeline, Gates = new GateEvaluator( - instance.WorkflowName, instance.WorkflowVersion, gates, _deps.GatePolicies, _deps.Approvals), + instance.WorkflowName, instance.WorkflowVersion, instance.TenantId, gates, + _deps.GatePolicies, _deps.Approvals), Approvals = _deps.ApprovalService, Services = _deps.Services, ExecutorInvoked = async (executorId, superstep) => diff --git a/src/Abacus.Run/Persistence/InMemoryStores.cs b/src/Abacus.Run/Persistence/InMemoryStores.cs index 18b4873..3848e07 100644 --- a/src/Abacus.Run/Persistence/InMemoryStores.cs +++ b/src/Abacus.Run/Persistence/InMemoryStores.cs @@ -1,330 +1,364 @@ -using System.Collections.Concurrent; -using System.Runtime.CompilerServices; -using Abacus.Run.Abstractions; -using Abacus.Run.Core; - -namespace Abacus.Run.Persistence; - -public sealed class InMemoryEventStore : IEventStore -{ - private readonly ConcurrentDictionary> _events = new(StringComparer.Ordinal); - private readonly object _sync = new(); - - public ValueTask AppendBatchAsync(IReadOnlyList events, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(events); - - lock (_sync) - { - foreach (EventEnvelope envelope in events) - { - List list = _events.GetOrAdd(envelope.InstanceId, _ => []); - list.Add(envelope); - } - } - - return ValueTask.CompletedTask; - } - - public ValueTask> QueryAsync(EventQuery query, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(query); - - List snapshot = Snapshot(query.InstanceId); - - IEnumerable filtered = snapshot.Where(e => e.Sequence > query.FromExclusive); - if (query.ToInclusive is { } to) - { - filtered = filtered.Where(e => e.Sequence <= to); - } - if (query.Types is { Count: > 0 } types) - { - filtered = filtered.Where(e => types.Contains(e.EventType, StringComparer.Ordinal)); - } - - EventEnvelope[] ordered = filtered.OrderBy(e => e.Sequence).ToArray(); - EventEnvelope[] page = ordered.Take(Math.Clamp(query.Limit, 1, 1000)).ToArray(); - string? nextCursor = page.Length < ordered.Length ? page[^1].Sequence.ToString() : null; - - return ValueTask.FromResult(new Page(page, ordered.Length, nextCursor)); - } - - public async IAsyncEnumerable ReadAsync( - string instanceId, long fromExclusive, [EnumeratorCancellation] CancellationToken cancellationToken) - { - foreach (EventEnvelope envelope in Snapshot(instanceId).Where(e => e.Sequence > fromExclusive).OrderBy(e => e.Sequence)) - { - cancellationToken.ThrowIfCancellationRequested(); - yield return envelope; - } - - await ValueTask.CompletedTask; - } - - public ValueTask MaxSequenceAsync(string instanceId, CancellationToken cancellationToken) - { - List snapshot = Snapshot(instanceId); - return ValueTask.FromResult(snapshot.Count == 0 ? 0 : snapshot.Max(e => e.Sequence)); - } - - private List Snapshot(string instanceId) - { - lock (_sync) - { - return _events.TryGetValue(instanceId, out List? list) ? [.. list] : []; - } - } -} - -public sealed class InMemoryLogStore : ILogStore -{ - private readonly ConcurrentDictionary> _logs = new(StringComparer.Ordinal); - private readonly object _sync = new(); - - public ValueTask AppendAsync(InstanceLogEntry entry, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(entry); - lock (_sync) - { - _logs.GetOrAdd(entry.InstanceId, _ => []).Add(entry); - } - return ValueTask.CompletedTask; - } - - public ValueTask> QueryAsync( - string instanceId, string? level, string? executorId, int limit, CancellationToken cancellationToken) - { - List snapshot; - lock (_sync) - { - snapshot = _logs.TryGetValue(instanceId, out List? list) ? [.. list] : []; - } - - IEnumerable filtered = snapshot; - if (level is { Length: > 0 }) - { - filtered = filtered.Where(l => string.Equals(l.Level, level, StringComparison.OrdinalIgnoreCase)); - } - if (executorId is { Length: > 0 }) - { - filtered = filtered.Where(l => l.ExecutorId == executorId); - } - - return ValueTask.FromResult>( - filtered.OrderBy(l => l.LoggedAt).Take(Math.Clamp(limit, 1, 1000)).ToArray()); - } -} - -public sealed class InMemoryApprovalStore : IApprovalStore -{ - private readonly ConcurrentDictionary _approvals = new(StringComparer.Ordinal); - private readonly ConcurrentDictionary> _decisions = new(StringComparer.Ordinal); - private readonly object _sync = new(); - - public ValueTask CreateAsync(ApprovalRequest request, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(request); - if (!_approvals.TryAdd(request.ApprovalId, request)) - { - throw new InvalidOperationException($"Approval '{request.ApprovalId}' already exists."); - } - return ValueTask.FromResult(request); - } - - public ValueTask GetAsync(string approvalId, CancellationToken cancellationToken) - => ValueTask.FromResult(_approvals.TryGetValue(approvalId, out ApprovalRequest? a) ? a : null); - - public ValueTask> ListForInstanceAsync(string instanceId, CancellationToken cancellationToken) - => ValueTask.FromResult>( - _approvals.Values.Where(a => a.InstanceId == instanceId).OrderBy(a => a.CreatedAt).ToArray()); - - public ValueTask> QueryPendingAsync( - string? tenantId, IReadOnlyList? assignees, int limit, CancellationToken cancellationToken) - { - IEnumerable results = _approvals.Values.Where(a => a.State == ApprovalState.Pending); - - if (tenantId is { Length: > 0 }) - { - results = results.Where(a => a.TenantId == tenantId); - } - if (assignees is { Count: > 0 }) - { - results = results.Where(a => a.Assignees.Count == 0 || a.Assignees.Any(x => assignees.Contains(x))); - } - - // Expiring soonest first: the queue should surface what is about to time out. - return ValueTask.FromResult>( - results.OrderBy(a => a.ExpiresAt).Take(Math.Clamp(limit, 1, 500)).ToArray()); - } - - public ValueTask> ClaimExpiredAsync( - DateTimeOffset now, int max, CancellationToken cancellationToken) - { - lock (_sync) - { - ApprovalRequest[] due = _approvals.Values - .Where(a => a.State == ApprovalState.Pending && a.ExpiresAt <= now) - .OrderBy(a => a.ExpiresAt) - .Take(max) - .ToArray(); - - return ValueTask.FromResult>(due); - } - } - - public ValueTask TryRecordDecisionAsync( - string approvalId, ApprovalDecision decision, ApprovalState? newState, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(decision); - - lock (_sync) - { - if (!_approvals.TryGetValue(approvalId, out ApprovalRequest? current) || current.State != ApprovalState.Pending) - { - return ValueTask.FromResult(false); - } - - List votes = _decisions.GetOrAdd(approvalId, _ => []); - if (votes.Any(v => string.Equals(v.DeciderId, decision.DeciderId, StringComparison.OrdinalIgnoreCase))) - { - return ValueTask.FromResult(false); // one vote per decider - } - - votes.Add(decision); - - if (newState is { } state) - { - _approvals[approvalId] = current with { State = state }; - } - - return ValueTask.FromResult(true); - } - } - - public ValueTask> GetDecisionsAsync(string approvalId, CancellationToken cancellationToken) - { - lock (_sync) - { - return ValueTask.FromResult>( - _decisions.TryGetValue(approvalId, out List? votes) ? [.. votes] : []); - } - } - - public ValueTask TrySetStateAsync(string approvalId, ApprovalState state, CancellationToken cancellationToken) - { - lock (_sync) - { - if (!_approvals.TryGetValue(approvalId, out ApprovalRequest? current)) - { - return ValueTask.FromResult(false); - } - - _approvals[approvalId] = current with - { - State = state, - EscalatedOnce = current.EscalatedOnce || state == ApprovalState.Pending - }; - return ValueTask.FromResult(true); - } - } - - public ValueTask CancelForInstanceAsync(string instanceId, CancellationToken cancellationToken) - { - lock (_sync) - { - foreach (ApprovalRequest approval in _approvals.Values.Where(a => - a.InstanceId == instanceId && a.State == ApprovalState.Pending)) - { - _approvals[approval.ApprovalId] = approval with { State = ApprovalState.Cancelled }; - } - } - return ValueTask.CompletedTask; - } -} - -public sealed class InMemoryGatePolicyStore : IGatePolicyStore -{ - private readonly ConcurrentDictionary _policies = new(StringComparer.OrdinalIgnoreCase); - private readonly ConcurrentDictionary _overrides = new(StringComparer.OrdinalIgnoreCase); - - public ValueTask FindAsync( - string workflowName, string workflowVersion, string executorId, string? instanceId, CancellationToken cancellationToken) - { - // Per-instance override outranks the workflow-level policy. - if (instanceId is { Length: > 0 } && _overrides.TryGetValue($"{instanceId}|{executorId}", out ApprovalGate? instanceGate)) - { - return ValueTask.FromResult(instanceGate); - } - - return ValueTask.FromResult( - _policies.TryGetValue($"{workflowName}|{workflowVersion}|{executorId}", out ApprovalGate? gate) ? gate : null); - } - - public ValueTask SetAsync( - string workflowName, string workflowVersion, string executorId, ApprovalGate gate, CancellationToken cancellationToken) - { - _policies[$"{workflowName}|{workflowVersion}|{executorId}"] = gate; - return ValueTask.CompletedTask; - } - - public ValueTask SetInstanceOverrideAsync(string instanceId, string executorId, ApprovalGate gate, CancellationToken cancellationToken) - { - _overrides[$"{instanceId}|{executorId}"] = gate; - return ValueTask.CompletedTask; - } -} - -public sealed class InMemoryBlobStore : IBlobStore -{ - private readonly ConcurrentDictionary _blobs = new(StringComparer.Ordinal); - - public int Count => _blobs.Count; - - public ValueTask UploadAsync(string key, byte[] content, CancellationToken cancellationToken) - { - string uri = $"mem://{key}"; - _blobs[uri] = content; - return ValueTask.FromResult(uri); - } - - public ValueTask DownloadAsync(string uri, CancellationToken cancellationToken) - => _blobs.TryGetValue(uri, out byte[]? content) - ? ValueTask.FromResult(content) - : throw new KeyNotFoundException($"Blob '{uri}' not found."); - - public ValueTask DeleteAsync(string uri, CancellationToken cancellationToken) - { - _blobs.TryRemove(uri, out _); - return ValueTask.CompletedTask; - } -} - -public sealed class InMemoryAuditStore : IAuditStore -{ - private readonly List _entries = []; - private readonly object _sync = new(); - - public ValueTask WriteAsync(AuditEntry entry, CancellationToken cancellationToken) - { - ArgumentNullException.ThrowIfNull(entry); - lock (_sync) - { - _entries.Add(entry); - } - return ValueTask.CompletedTask; - } - - public ValueTask> QueryAsync(string? instanceId, int limit, CancellationToken cancellationToken) - { - lock (_sync) - { - IEnumerable results = _entries; - if (instanceId is { Length: > 0 }) - { - results = results.Where(e => e.InstanceId == instanceId); - } - return ValueTask.FromResult>( - results.OrderByDescending(e => e.OccurredAt).Take(Math.Clamp(limit, 1, 1000)).ToArray()); - } - } -} +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +using Abacus.Run.Abstractions; +using Abacus.Run.Core; + +namespace Abacus.Run.Persistence; + +public sealed class InMemoryEventStore : IEventStore +{ + private readonly ConcurrentDictionary> _events = new(StringComparer.Ordinal); + private readonly object _sync = new(); + + public ValueTask AppendBatchAsync(IReadOnlyList events, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(events); + + lock (_sync) + { + foreach (EventEnvelope envelope in events) + { + List list = _events.GetOrAdd(envelope.InstanceId, _ => []); + list.Add(envelope); + } + } + + return ValueTask.CompletedTask; + } + + public ValueTask> QueryAsync(EventQuery query, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(query); + + List snapshot = Snapshot(query.InstanceId); + + IEnumerable filtered = snapshot.Where(e => e.Sequence > query.FromExclusive); + if (query.ToInclusive is { } to) + { + filtered = filtered.Where(e => e.Sequence <= to); + } + if (query.Types is { Count: > 0 } types) + { + filtered = filtered.Where(e => types.Contains(e.EventType, StringComparer.Ordinal)); + } + + EventEnvelope[] ordered = filtered.OrderBy(e => e.Sequence).ToArray(); + EventEnvelope[] page = ordered.Take(Math.Clamp(query.Limit, 1, 1000)).ToArray(); + string? nextCursor = page.Length < ordered.Length ? page[^1].Sequence.ToString() : null; + + return ValueTask.FromResult(new Page(page, ordered.Length, nextCursor)); + } + + public async IAsyncEnumerable ReadAsync( + string instanceId, long fromExclusive, [EnumeratorCancellation] CancellationToken cancellationToken) + { + foreach (EventEnvelope envelope in Snapshot(instanceId).Where(e => e.Sequence > fromExclusive).OrderBy(e => e.Sequence)) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return envelope; + } + + await ValueTask.CompletedTask; + } + + public ValueTask MaxSequenceAsync(string instanceId, CancellationToken cancellationToken) + { + List snapshot = Snapshot(instanceId); + return ValueTask.FromResult(snapshot.Count == 0 ? 0 : snapshot.Max(e => e.Sequence)); + } + + private List Snapshot(string instanceId) + { + lock (_sync) + { + return _events.TryGetValue(instanceId, out List? list) ? [.. list] : []; + } + } +} + +public sealed class InMemoryLogStore : ILogStore +{ + private readonly ConcurrentDictionary> _logs = new(StringComparer.Ordinal); + private readonly object _sync = new(); + + public ValueTask AppendAsync(InstanceLogEntry entry, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(entry); + lock (_sync) + { + _logs.GetOrAdd(entry.InstanceId, _ => []).Add(entry); + } + return ValueTask.CompletedTask; + } + + public ValueTask> QueryAsync( + string instanceId, string? level, string? executorId, int limit, CancellationToken cancellationToken) + { + List snapshot; + lock (_sync) + { + snapshot = _logs.TryGetValue(instanceId, out List? list) ? [.. list] : []; + } + + IEnumerable filtered = snapshot; + if (level is { Length: > 0 }) + { + filtered = filtered.Where(l => string.Equals(l.Level, level, StringComparison.OrdinalIgnoreCase)); + } + if (executorId is { Length: > 0 }) + { + filtered = filtered.Where(l => l.ExecutorId == executorId); + } + + return ValueTask.FromResult>( + filtered.OrderBy(l => l.LoggedAt).Take(Math.Clamp(limit, 1, 1000)).ToArray()); + } +} + +public sealed class InMemoryApprovalStore : IApprovalStore +{ + private readonly ConcurrentDictionary _approvals = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _decisions = new(StringComparer.Ordinal); + private readonly object _sync = new(); + + public ValueTask CreateAsync(ApprovalRequest request, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(request); + if (!_approvals.TryAdd(request.ApprovalId, request)) + { + throw new InvalidOperationException($"Approval '{request.ApprovalId}' already exists."); + } + return ValueTask.FromResult(request); + } + + public ValueTask GetAsync(string approvalId, CancellationToken cancellationToken) + => ValueTask.FromResult(_approvals.TryGetValue(approvalId, out ApprovalRequest? a) ? a : null); + + public ValueTask> ListForInstanceAsync(string instanceId, CancellationToken cancellationToken) + => ValueTask.FromResult>( + _approvals.Values.Where(a => a.InstanceId == instanceId).OrderBy(a => a.CreatedAt).ToArray()); + + public ValueTask> QueryPendingAsync( + string? tenantId, IReadOnlyList? assignees, int limit, CancellationToken cancellationToken) + { + IEnumerable results = _approvals.Values.Where(a => a.State == ApprovalState.Pending); + + if (tenantId is { Length: > 0 }) + { + results = results.Where(a => a.TenantId == tenantId); + } + if (assignees is { Count: > 0 }) + { + results = results.Where(a => a.Assignees.Count == 0 || a.Assignees.Any(x => assignees.Contains(x))); + } + + // Expiring soonest first: the queue should surface what is about to time out. + return ValueTask.FromResult>( + results.OrderBy(a => a.ExpiresAt).Take(Math.Clamp(limit, 1, 500)).ToArray()); + } + + public ValueTask> ClaimExpiredAsync( + DateTimeOffset now, int max, CancellationToken cancellationToken) + { + lock (_sync) + { + ApprovalRequest[] due = _approvals.Values + .Where(a => a.State == ApprovalState.Pending && a.ExpiresAt <= now) + .OrderBy(a => a.ExpiresAt) + .Take(max) + .ToArray(); + + return ValueTask.FromResult>(due); + } + } + + public ValueTask TryRecordDecisionAsync( + string approvalId, ApprovalDecision decision, ApprovalState? newState, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(decision); + + lock (_sync) + { + if (!_approvals.TryGetValue(approvalId, out ApprovalRequest? current) || current.State != ApprovalState.Pending) + { + return ValueTask.FromResult(false); + } + + List votes = _decisions.GetOrAdd(approvalId, _ => []); + if (votes.Any(v => string.Equals(v.DeciderId, decision.DeciderId, StringComparison.OrdinalIgnoreCase))) + { + return ValueTask.FromResult(false); // one vote per decider + } + + votes.Add(decision); + + if (newState is { } state) + { + _approvals[approvalId] = current with { State = state }; + } + + return ValueTask.FromResult(true); + } + } + + public ValueTask> GetDecisionsAsync(string approvalId, CancellationToken cancellationToken) + { + lock (_sync) + { + return ValueTask.FromResult>( + _decisions.TryGetValue(approvalId, out List? votes) ? [.. votes] : []); + } + } + + public ValueTask TrySetStateAsync(string approvalId, ApprovalState state, CancellationToken cancellationToken) + { + lock (_sync) + { + if (!_approvals.TryGetValue(approvalId, out ApprovalRequest? current)) + { + return ValueTask.FromResult(false); + } + + _approvals[approvalId] = current with + { + State = state, + EscalatedOnce = current.EscalatedOnce || state == ApprovalState.Pending + }; + return ValueTask.FromResult(true); + } + } + + public ValueTask CancelForInstanceAsync(string instanceId, CancellationToken cancellationToken) + { + lock (_sync) + { + foreach (ApprovalRequest approval in _approvals.Values.Where(a => + a.InstanceId == instanceId && a.State == ApprovalState.Pending)) + { + _approvals[approval.ApprovalId] = approval with { State = ApprovalState.Cancelled }; + } + } + return ValueTask.CompletedTask; + } +} + +public sealed class InMemoryGatePolicyStore : IGatePolicyStore +{ + private readonly ConcurrentDictionary _policies = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary _overrides = new(StringComparer.OrdinalIgnoreCase); + + /// Host-wide policies are stored under a scope no tenant id can collide with. + private const string HostScope = "host"; + + public ValueTask FindAsync( + string? tenantId, string workflowName, string workflowVersion, string executorId, string? instanceId, + CancellationToken cancellationToken) + { + // Per-instance override outranks every policy scope. + if (instanceId is { Length: > 0 } && _overrides.TryGetValue($"{instanceId}|{executorId}", out ApprovalGate? instanceGate)) + { + return ValueTask.FromResult(instanceGate); + } + + if (tenantId is { Length: > 0 } && + _policies.TryGetValue(Key(tenantId, workflowName, workflowVersion, executorId), out ApprovalGate? tenantGate)) + { + return ValueTask.FromResult(tenantGate); + } + + return ValueTask.FromResult( + _policies.TryGetValue(Key(null, workflowName, workflowVersion, executorId), out ApprovalGate? gate) ? gate : null); + } + + public ValueTask> ListAsync( + string? tenantId, string workflowName, string workflowVersion, CancellationToken cancellationToken) + { + string prefix = ScopePrefix(tenantId, workflowName, workflowVersion); + + Dictionary matches = _policies + .Where(entry => entry.Key.StartsWith(prefix, StringComparison.OrdinalIgnoreCase)) + .ToDictionary(entry => entry.Key[prefix.Length..], entry => entry.Value, StringComparer.Ordinal); + + return ValueTask.FromResult>(matches); + } + + public ValueTask SetAsync( + string? tenantId, string workflowName, string workflowVersion, string executorId, ApprovalGate gate, + CancellationToken cancellationToken) + { + _policies[Key(tenantId, workflowName, workflowVersion, executorId)] = gate; + return ValueTask.CompletedTask; + } + + public ValueTask RemoveAsync( + string? tenantId, string workflowName, string workflowVersion, string executorId, + CancellationToken cancellationToken) + => ValueTask.FromResult(_policies.TryRemove(Key(tenantId, workflowName, workflowVersion, executorId), out _)); + + public ValueTask SetInstanceOverrideAsync(string instanceId, string executorId, ApprovalGate gate, CancellationToken cancellationToken) + { + _overrides[$"{instanceId}|{executorId}"] = gate; + return ValueTask.CompletedTask; + } + + private static string ScopePrefix(string? tenantId, string workflowName, string workflowVersion) + => $"{(string.IsNullOrEmpty(tenantId) ? HostScope : tenantId)}|{workflowName}|{workflowVersion}|"; + + private static string Key(string? tenantId, string workflowName, string workflowVersion, string executorId) + => ScopePrefix(tenantId, workflowName, workflowVersion) + executorId; +} + +public sealed class InMemoryBlobStore : IBlobStore +{ + private readonly ConcurrentDictionary _blobs = new(StringComparer.Ordinal); + + public int Count => _blobs.Count; + + public ValueTask UploadAsync(string key, byte[] content, CancellationToken cancellationToken) + { + string uri = $"mem://{key}"; + _blobs[uri] = content; + return ValueTask.FromResult(uri); + } + + public ValueTask DownloadAsync(string uri, CancellationToken cancellationToken) + => _blobs.TryGetValue(uri, out byte[]? content) + ? ValueTask.FromResult(content) + : throw new KeyNotFoundException($"Blob '{uri}' not found."); + + public ValueTask DeleteAsync(string uri, CancellationToken cancellationToken) + { + _blobs.TryRemove(uri, out _); + return ValueTask.CompletedTask; + } +} + +public sealed class InMemoryAuditStore : IAuditStore +{ + private readonly List _entries = []; + private readonly object _sync = new(); + + public ValueTask WriteAsync(AuditEntry entry, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(entry); + lock (_sync) + { + _entries.Add(entry); + } + return ValueTask.CompletedTask; + } + + public ValueTask> QueryAsync(string? instanceId, int limit, CancellationToken cancellationToken) + { + lock (_sync) + { + IEnumerable results = _entries; + if (instanceId is { Length: > 0 }) + { + results = results.Where(e => e.InstanceId == instanceId); + } + return ValueTask.FromResult>( + results.OrderByDescending(e => e.OccurredAt).Take(Math.Clamp(limit, 1, 1000)).ToArray()); + } + } +} diff --git a/tests/Abacus.Run.IntegrationTests/HostFixture.cs b/tests/Abacus.Run.IntegrationTests/HostFixture.cs index f715441..3fb8215 100644 --- a/tests/Abacus.Run.IntegrationTests/HostFixture.cs +++ b/tests/Abacus.Run.IntegrationTests/HostFixture.cs @@ -177,6 +177,78 @@ protected override ValueTask ExecuteCoreAsync( } } +/// +/// Three nodes with nothing gated by default, so a tenant's configuration is the only thing that +/// can make the run stop — except settle, which the author locked. +/// +public sealed class TenantOrderWorkflow : IWorkflowDefinition +{ + private readonly SideEffectLedger _ledger; + + public TenantOrderWorkflow(SideEffectLedger ledger) => _ledger = ledger; + + public string Name => "tenant-order"; + public string Version => "1.0.0"; + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + ExecutorBinding prepare = context.Node(new Step("prepare", _ledger, context.InstanceId)); + ExecutorBinding dispatch = context.Node(new Step("dispatch", _ledger, context.InstanceId)); + + ExecutorBinding settle = context.Node( + new Settle("settle", _ledger, context.InstanceId), + gate => gate + .When(order => order.Amount > 25_000m) + .Reason("RegulatedSettlement") + .Locked()); + + return new ValueTask(new WorkflowBuilder(prepare) + .AddEdge(prepare, dispatch) + .AddEdge(dispatch, settle) + .WithOutputFrom(settle) + .WithName(Name) + .Build()); + } + + private sealed class Step : HostExecutor + { + private readonly SideEffectLedger _ledger; + private readonly string _instanceId; + + public Step(string id, SideEffectLedger ledger, string instanceId) : base(id) + { + _ledger = ledger; + _instanceId = instanceId; + } + + protected override ValueTask ExecuteCoreAsync( + OrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + _ledger.Record(Id, _instanceId); + return ValueTask.FromResult(input); + } + } + + private sealed class Settle : HostExecutor + { + private readonly SideEffectLedger _ledger; + private readonly string _instanceId; + + public Settle(string id, SideEffectLedger ledger, string instanceId) : base(id) + { + _ledger = ledger; + _instanceId = instanceId; + } + + protected override ValueTask ExecuteCoreAsync( + OrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + _ledger.Record(Id, _instanceId); + return ValueTask.FromResult(new OrderResult(input.OrderId, "settled")); + } + } +} + public sealed class HostFixture : WebApplicationFactory { public SideEffectLedger Ledger { get; } = new(); @@ -188,6 +260,7 @@ protected override IHost CreateHost(IHostBuilder builder) services.AddSingleton(Ledger); services.AddSingleton(sp => new OrderWorkflow(sp.GetRequiredService())); services.AddSingleton(sp => new GatedOrderWorkflow(sp.GetRequiredService())); + services.AddSingleton(sp => new TenantOrderWorkflow(sp.GetRequiredService())); }); return base.CreateHost(builder); @@ -230,7 +303,7 @@ public async Task WaitForStatusAsync( } public async Task StartAsync( - HttpClient client, string workflow, object context, string? idempotencyKey = null) + HttpClient client, string workflow, object context, string? idempotencyKey = null, string? tenantId = null) { using var request = new HttpRequestMessage(HttpMethod.Post, $"/workflows/{workflow}/instances") { @@ -242,6 +315,11 @@ public async Task StartAsync( request.Headers.Add("Idempotency-Key", idempotencyKey); } + if (tenantId is not null) + { + request.Headers.Add("X-Tenant-Id", tenantId); + } + HttpResponseMessage response = await client.SendAsync(request); response.EnsureSuccessStatusCode(); diff --git a/tests/Abacus.Run.IntegrationTests/TenantGateConfigurationTests.cs b/tests/Abacus.Run.IntegrationTests/TenantGateConfigurationTests.cs new file mode 100644 index 0000000..2dd774a --- /dev/null +++ b/tests/Abacus.Run.IntegrationTests/TenantGateConfigurationTests.cs @@ -0,0 +1,343 @@ +using System.Net; +using System.Net.Http.Json; +using System.Text.Json; +using Abacus.Run.Abstractions; +using FluentAssertions; +using Xunit; + +namespace Abacus.Run.IntegrationTests; + +/// +/// The configuration API itself: what a tenant can see, change, and is refused. +/// +public class NodeConfigurationApiTests : IClassFixture +{ + private readonly HostFixture _host; + + public NodeConfigurationApiTests(HostFixture host) => _host = host; + + private HttpClient ClientFor(string tenantId) + { + HttpClient client = _host.CreateClient(); + client.DefaultRequestHeaders.Add("X-Tenant-Id", tenantId); + return client; + } + + private static JsonElement Node(JsonElement body, string executorId) + => body.GetProperty("nodes").EnumerateArray() + .Single(n => n.GetProperty("executorId").GetString() == executorId); + + [Fact] + public async Task The_catalog_lists_every_executor_node_of_a_version() + { + HttpClient client = ClientFor("cat-tenant"); + + JsonElement body = await client.GetFromJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes"); + + body.GetProperty("workflowName").GetString().Should().Be("tenant-order"); + body.GetProperty("workflowVersion").GetString().Should().Be("1.0.0"); + body.GetProperty("tenantId").GetString().Should().Be("cat-tenant"); + + body.GetProperty("nodes").EnumerateArray() + .Select(n => n.GetProperty("executorId").GetString()) + .Should().Equal("prepare", "dispatch", "settle"); + } + + [Fact] + public async Task Nodes_the_author_did_not_gate_are_autonomous_by_default() + { + JsonElement body = await ClientFor("default-tenant") + .GetFromJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes"); + + JsonElement dispatch = Node(body, "dispatch"); + dispatch.GetProperty("effective").GetProperty("mode").GetString().Should().Be("autonomous"); + dispatch.GetProperty("effectiveSource").GetString().Should().Be("definition"); + dispatch.GetProperty("tenantOverride").ValueKind.Should().Be(JsonValueKind.Null); + dispatch.GetProperty("locked").GetBoolean().Should().BeFalse(); + } + + [Fact] + public async Task A_locked_node_is_advertised_as_locked_with_its_declared_gate() + { + JsonElement body = await ClientFor("locked-tenant") + .GetFromJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes"); + + JsonElement settle = Node(body, "settle"); + settle.GetProperty("locked").GetBoolean().Should().BeTrue(); + settle.GetProperty("declared").GetProperty("mode").GetString().Should().Be("conditional"); + settle.GetProperty("declared").GetProperty("reason").GetString().Should().Be("RegulatedSettlement"); + } + + [Fact] + public async Task Configuring_a_node_is_reflected_in_the_catalog_for_that_tenant_only() + { + HttpClient owner = ClientFor("api-owner"); + HttpClient bystander = ClientFor("api-bystander"); + + HttpResponseMessage put = await owner.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", + new { mode = "requireApproval", reason = "four eyes", assignees = new[] { "group:ops" } }); + + put.StatusCode.Should().Be(HttpStatusCode.OK); + + JsonElement dispatch = Node(await put.Content.ReadFromJsonAsync(), "dispatch"); + dispatch.GetProperty("effective").GetProperty("mode").GetString().Should().Be("requireApproval"); + dispatch.GetProperty("effectiveSource").GetString().Should().Be("tenant"); + dispatch.GetProperty("tenantOverride").GetProperty("reason").GetString().Should().Be("four eyes"); + + JsonElement theirs = await bystander.GetFromJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes"); + Node(theirs, "dispatch").GetProperty("effective").GetProperty("mode").GetString().Should().Be("autonomous"); + } + + [Fact] + public async Task Several_nodes_can_be_configured_in_one_call() + { + HttpClient client = ClientFor("api-bulk"); + + HttpResponseMessage put = await client.PutAsJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes", new + { + nodes = new Dictionary + { + ["prepare"] = new { mode = "requireApproval" }, + ["dispatch"] = new { mode = "requireApproval" } + } + }); + + put.StatusCode.Should().Be(HttpStatusCode.OK); + + JsonElement body = await put.Content.ReadFromJsonAsync(); + Node(body, "prepare").GetProperty("effectiveSource").GetString().Should().Be("tenant"); + Node(body, "dispatch").GetProperty("effectiveSource").GetString().Should().Be("tenant"); + } + + [Fact] + public async Task Deleting_a_configuration_restores_the_declared_gate() + { + HttpClient client = ClientFor("api-reset"); + + await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", new { mode = "requireApproval" }); + + HttpResponseMessage delete = await client.DeleteAsync("/workflows/tenant-order/versions/1.0.0/nodes/dispatch"); + delete.StatusCode.Should().Be(HttpStatusCode.OK); + + JsonElement dispatch = Node(await delete.Content.ReadFromJsonAsync(), "dispatch"); + dispatch.GetProperty("effective").GetProperty("mode").GetString().Should().Be("autonomous"); + dispatch.GetProperty("tenantOverride").ValueKind.Should().Be(JsonValueKind.Null); + } + + [Fact] + public async Task Un_gating_a_locked_node_is_a_conflict() + { + HttpClient client = ClientFor("api-locked"); + + HttpResponseMessage put = await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/settle", new { mode = "autonomous" }); + + put.StatusCode.Should().Be(HttpStatusCode.Conflict); + + JsonElement body = await client.GetFromJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes"); + Node(body, "settle").GetProperty("tenantOverride").ValueKind + .Should().Be(JsonValueKind.Null, "a refused write must persist nothing"); + } + + [Fact] + public async Task Tightening_a_locked_node_is_allowed() + { + HttpClient client = ClientFor("api-tighten"); + + HttpResponseMessage put = await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/settle", + new { mode = "requireApproval", requiredApprovers = 2 }); + + put.StatusCode.Should().Be(HttpStatusCode.OK); + + JsonElement settle = Node(await put.Content.ReadFromJsonAsync(), "settle"); + settle.GetProperty("effective").GetProperty("mode").GetString().Should().Be("requireApproval"); + settle.GetProperty("effective").GetProperty("requiredApprovers").GetInt32().Should().Be(2); + } + + [Fact] + public async Task An_unknown_workflow_version_or_executor_is_not_found() + { + HttpClient client = ClientFor("api-404"); + + (await client.GetAsync("/workflows/nope/versions/1.0.0/nodes")) + .StatusCode.Should().Be(HttpStatusCode.NotFound); + (await client.GetAsync("/workflows/tenant-order/versions/9.9.9/nodes")) + .StatusCode.Should().Be(HttpStatusCode.NotFound); + (await client.PutAsJsonAsync("/workflows/tenant-order/versions/1.0.0/nodes/ghost", new { mode = "autonomous" })) + .StatusCode.Should().Be(HttpStatusCode.NotFound); + } + + [Fact] + public async Task An_unsupported_mode_is_a_validation_problem() + { + HttpClient client = ClientFor("api-invalid"); + + HttpResponseMessage put = await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", new { mode = "conditional" }); + + put.StatusCode.Should().Be(HttpStatusCode.BadRequest, "a predicate cannot be supplied over HTTP"); + } +} + +/// +/// The point of the whole feature: what the runtime does with a tenant's configuration. +/// +public class TenantGatedExecutionTests : IClassFixture +{ + private readonly HostFixture _host; + + public TenantGatedExecutionTests(HostFixture host) => _host = host; + + private HttpClient ClientFor(string tenantId) + { + HttpClient client = _host.CreateClient(); + client.DefaultRequestHeaders.Add("X-Tenant-Id", tenantId); + return client; + } + + [Fact] + public async Task With_no_configuration_every_node_runs_autonomously() + { + HttpClient client = ClientFor("run-default"); + + string instanceId = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-AUTO", Amount: 100m), tenantId: "run-default"); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + _host.Ledger.Entries.Should().Contain($"dispatch:{instanceId}"); + _host.Ledger.Entries.Should().Contain($"settle:{instanceId}"); + } + + [Fact] + public async Task A_tenant_that_requires_approval_parks_before_the_side_effect() + { + HttpClient client = ClientFor("run-gated"); + + HttpResponseMessage put = await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", + new { mode = "requireApproval", reason = "TenantPolicy", assignees = new[] { "group:ops" } }); + put.EnsureSuccessStatusCode(); + + string instanceId = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-GATED", Amount: 100m), tenantId: "run-gated"); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.AwaitingApproval); + + _host.Ledger.Entries.Should().Contain($"prepare:{instanceId}"); + _host.Ledger.Entries.Should().NotContain($"dispatch:{instanceId}", + "the tenant's gate must stop the node before its side effect"); + + JsonElement approvals = await client.GetFromJsonAsync($"/instances/{instanceId}/approvals"); + JsonElement approval = approvals.GetProperty("items").EnumerateArray().Single(); + approval.GetProperty("executorId").GetString().Should().Be("dispatch"); + approval.GetProperty("reason").GetString().Should().Be("TenantPolicy"); + approval.GetProperty("assignees").EnumerateArray().Select(a => a.GetString()).Should().Contain("group:ops"); + } + + [Fact] + public async Task Approving_a_tenant_configured_gate_resumes_the_run() + { + HttpClient client = ClientFor("run-approve"); + + (await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", + new { mode = "requireApproval" })).EnsureSuccessStatusCode(); + + string instanceId = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-RESUME", Amount: 100m), tenantId: "run-approve"); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.AwaitingApproval); + + JsonElement approvals = await client.GetFromJsonAsync($"/instances/{instanceId}/approvals"); + string approvalId = approvals.GetProperty("items").EnumerateArray().Single() + .GetProperty("approvalId").GetString()!; + + (await client.PostAsJsonAsync($"/approvals/{approvalId}/decision", new { decision = "approve" })) + .StatusCode.Should().Be(HttpStatusCode.Accepted); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + _host.Ledger.Entries.Count(e => e == $"dispatch:{instanceId}") + .Should().Be(1, "the approved node runs exactly once"); + } + + [Fact] + public async Task One_tenants_gate_does_not_stop_another_tenants_run() + { + HttpClient cautious = ClientFor("run-cautious"); + (await cautious.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", + new { mode = "requireApproval" })).EnsureSuccessStatusCode(); + + string gated = await _host.StartAsync( + cautious, "tenant-order", new OrderContext("ORD-CAUTIOUS", Amount: 100m), tenantId: "run-cautious"); + string free = await _host.StartAsync( + ClientFor("run-relaxed"), "tenant-order", new OrderContext("ORD-RELAXED", Amount: 100m), + tenantId: "run-relaxed"); + + await _host.WaitForStatusAsync(gated, InstanceStatus.AwaitingApproval); + await _host.WaitForStatusAsync(free, InstanceStatus.Completed); + + _host.Ledger.Entries.Should().NotContain($"dispatch:{gated}"); + _host.Ledger.Entries.Should().Contain($"dispatch:{free}"); + } + + [Fact] + public async Task Removing_the_configuration_lets_the_next_run_go_through() + { + HttpClient client = ClientFor("run-reset"); + + (await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/dispatch", + new { mode = "requireApproval" })).EnsureSuccessStatusCode(); + + string parked = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-BEFORE", Amount: 100m), tenantId: "run-reset"); + await _host.WaitForStatusAsync(parked, InstanceStatus.AwaitingApproval); + + (await client.DeleteAsync("/workflows/tenant-order/versions/1.0.0/nodes/dispatch")).EnsureSuccessStatusCode(); + + string after = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-AFTER", Amount: 100m), tenantId: "run-reset"); + + await _host.WaitForStatusAsync(after, InstanceStatus.Completed); + _host.Ledger.Entries.Should().Contain($"dispatch:{after}"); + } + + [Fact] + public async Task A_locked_gate_still_stops_a_tenant_that_tried_to_remove_it() + { + HttpClient client = ClientFor("run-locked"); + + (await client.PutAsJsonAsync( + "/workflows/tenant-order/versions/1.0.0/nodes/settle", new { mode = "autonomous" })) + .StatusCode.Should().Be(HttpStatusCode.Conflict); + + string instanceId = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-LOCKED", Amount: 50_000m), tenantId: "run-locked"); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.AwaitingApproval); + + _host.Ledger.Entries.Should().NotContain($"settle:{instanceId}"); + + JsonElement approvals = await client.GetFromJsonAsync($"/instances/{instanceId}/approvals"); + approvals.GetProperty("items").EnumerateArray().Single() + .GetProperty("executorId").GetString().Should().Be("settle"); + } + + [Fact] + public async Task A_locked_conditional_gate_still_lets_a_below_threshold_run_through() + { + HttpClient client = ClientFor("run-below"); + + string instanceId = await _host.StartAsync( + client, "tenant-order", new OrderContext("ORD-BELOW", Amount: 10m), tenantId: "run-below"); + + await _host.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + _host.Ledger.Entries.Should().Contain($"settle:{instanceId}"); + } +} diff --git a/tests/Abacus.Run.UnitTests/CheckpointAndStoreTests.cs b/tests/Abacus.Run.UnitTests/CheckpointAndStoreTests.cs index a8e9d7a..d350df1 100644 --- a/tests/Abacus.Run.UnitTests/CheckpointAndStoreTests.cs +++ b/tests/Abacus.Run.UnitTests/CheckpointAndStoreTests.cs @@ -483,11 +483,42 @@ public async Task Gate_policy_store_round_trips_workflow_and_instance_scopes() var store = new InMemoryGatePolicyStore(); var gate = new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "policy" }; - await store.SetAsync("wf", "1.0.0", "pay", gate, default); + await store.SetAsync(null, "wf", "1.0.0", "pay", gate, default); - (await store.FindAsync("wf", "1.0.0", "pay", null, default))!.Reason.Should().Be("policy"); - (await store.FindAsync("wf", "2.0.0", "pay", null, default)).Should().BeNull("policies are version-scoped"); - (await store.FindAsync("wf", "1.0.0", "other", null, default)).Should().BeNull(); + (await store.FindAsync(null, "wf", "1.0.0", "pay", null, default))!.Reason.Should().Be("policy"); + (await store.FindAsync(null, "wf", "2.0.0", "pay", null, default)).Should().BeNull("policies are version-scoped"); + (await store.FindAsync(null, "wf", "1.0.0", "other", null, default)).Should().BeNull(); + } + + [Fact] + public async Task Gate_policy_store_scopes_by_tenant_and_lists_one_scope_at_a_time() + { + var store = new InMemoryGatePolicyStore(); + var gated = new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "tenant" }; + + await store.SetAsync(null, "wf", "1.0.0", "pay", new ApprovalGate { Reason = "host" }, default); + await store.SetAsync("t1", "wf", "1.0.0", "pay", gated, default); + + (await store.FindAsync("t1", "wf", "1.0.0", "pay", null, default))!.Reason.Should().Be("tenant"); + (await store.FindAsync("t2", "wf", "1.0.0", "pay", null, default))!.Reason + .Should().Be("host", "a tenant with no policy of its own falls back to the host scope"); + + (await store.ListAsync("t1", "wf", "1.0.0", default)).Should().ContainKey("pay"); + (await store.ListAsync("t2", "wf", "1.0.0", default)).Should().BeEmpty(); + (await store.ListAsync(null, "wf", "1.0.0", default)).Should().ContainKey("pay"); + } + + [Fact] + public async Task Gate_policy_removal_falls_back_to_the_next_scope() + { + var store = new InMemoryGatePolicyStore(); + await store.SetAsync(null, "wf", "1.0.0", "pay", new ApprovalGate { Reason = "host" }, default); + await store.SetAsync("t1", "wf", "1.0.0", "pay", new ApprovalGate { Reason = "tenant" }, default); + + (await store.RemoveAsync("t1", "wf", "1.0.0", "pay", default)).Should().BeTrue(); + (await store.RemoveAsync("t1", "wf", "1.0.0", "pay", default)).Should().BeFalse("it is already gone"); + + (await store.FindAsync("t1", "wf", "1.0.0", "pay", null, default))!.Reason.Should().Be("host"); } private static InstanceLogEntry Entry(string instanceId, string level, string executorId) => new() diff --git a/tests/Abacus.Run.UnitTests/GateConfigurationTests.cs b/tests/Abacus.Run.UnitTests/GateConfigurationTests.cs new file mode 100644 index 0000000..e289871 --- /dev/null +++ b/tests/Abacus.Run.UnitTests/GateConfigurationTests.cs @@ -0,0 +1,571 @@ +using Abacus.Run.Abstractions; +using Abacus.Run.Api; +using Abacus.Run.Core; +using Abacus.Run.Persistence; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows; +using Xunit; + +namespace Abacus.Run.UnitTests; + +/// +/// Three nodes: one plain (autonomous by default), one the author gated but left open to tenants, +/// one the author gated and locked. +/// +public sealed class ConfigurableWorkflow : IWorkflowDefinition +{ + public ConfigurableWorkflow(string name = "configurable", string version = "1.0.0") + { + Name = name; + Version = version; + } + + public string Name { get; } + public string Version { get; } + + public int Builds { get; private set; } + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + Builds++; + + ExecutorBinding validate = context.Node(new PassThrough("validate")); + + ExecutorBinding notify = context.Node(new PassThrough("notify"), gate => gate + .Mode(ExecutionMode.RequireApproval) + .Reason("NotifiesCustomers") + .AssignTo("group:ops") + .ExpiresAfter(TimeSpan.FromHours(4))); + + ExecutorBinding pay = context.Node(new Terminal("pay"), gate => gate + .When(c => c.Amount > 25_000m) + .Reason("AmountAboveThreshold") + .RequireApprovers(2) + .RequireSegregationOfDuties() + .Locked()); + + return new ValueTask(new WorkflowBuilder(validate) + .AddEdge(validate, notify) + .AddEdge(notify, pay) + .WithOutputFrom(pay) + .WithName(Name) + .Build()); + } + + private sealed class PassThrough : HostExecutor + { + public PassThrough(string id) : base(id) { } + + protected override ValueTask ExecuteCoreAsync( + SampleContext input, IWorkflowContext context, CancellationToken cancellationToken) + => ValueTask.FromResult(input); + } + + private sealed class Terminal : HostExecutor + { + public Terminal(string id) : base(id) { } + + protected override ValueTask ExecuteCoreAsync( + SampleContext input, IWorkflowContext context, CancellationToken cancellationToken) + => ValueTask.FromResult(new SampleResult(input.Value)); + } +} + +public class WorkflowInspectorTests +{ + [Fact] + public async Task Lists_every_declared_node_with_its_types() + { + IReadOnlyList nodes = await Inspect(new ConfigurableWorkflow()); + + nodes.Select(n => n.ExecutorId).Should().Equal("validate", "notify", "pay"); + nodes[0].InputType.Should().Be(nameof(SampleContext)); + nodes[2].OutputType.Should().Be(nameof(SampleResult)); + nodes.Should().OnlyContain(n => n.Configurable); + } + + [Fact] + public async Task A_node_declared_without_a_gate_is_autonomous() + { + IReadOnlyList nodes = await Inspect(new ConfigurableWorkflow()); + + nodes.Single(n => n.ExecutorId == "validate").DeclaredGate.Mode + .Should().Be(ExecutionMode.Autonomous, "autonomous is the host default for an ungated node"); + } + + [Fact] + public async Task Declared_gates_are_reported_verbatim() + { + IReadOnlyList nodes = await Inspect(new ConfigurableWorkflow()); + + WorkflowNodeDescriptor notify = nodes.Single(n => n.ExecutorId == "notify"); + notify.DeclaredGate.Mode.Should().Be(ExecutionMode.RequireApproval); + notify.DeclaredGate.Assignees.Should().Equal("group:ops"); + notify.DeclaredGate.Expiry.Should().Be(TimeSpan.FromHours(4)); + notify.DeclaredGate.Locked.Should().BeFalse(); + + WorkflowNodeDescriptor pay = nodes.Single(n => n.ExecutorId == "pay"); + pay.DeclaredGate.Mode.Should().Be(ExecutionMode.Conditional); + pay.DeclaredGate.Locked.Should().BeTrue(); + } + + [Fact] + public async Task A_version_is_built_only_once() + { + var definition = new ConfigurableWorkflow(); + var inspector = new WorkflowInspector(); + var descriptor = new WorkflowDescriptor(definition); + + await inspector.InspectAsync(descriptor, default); + await inspector.InspectAsync(descriptor, default); + + definition.Builds.Should().Be(1, "the graph of a registered version cannot change"); + } + + [Fact] + public async Task Raw_nodes_are_listed_but_not_configurable() + { + IReadOnlyList nodes = await Inspect(new RawNodeWorkflow()); + + nodes.Single(n => n.ExecutorId == "raw").Configurable + .Should().BeFalse("a raw binding runs outside the executor pipeline and cannot be gated"); + nodes.Single(n => n.ExecutorId == "start").Configurable.Should().BeTrue(); + } + + private static async Task> Inspect(IWorkflowDefinition definition) + => await new WorkflowInspector().InspectAsync(new WorkflowDescriptor(definition), default); + + private sealed class RawNodeWorkflow : IWorkflowDefinition + { + public string Name => "raw-workflow"; + public string Version => "1.0.0"; + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + ExecutorBinding start = context.Node(new Start("start")); + ExecutorBinding raw = context.RawNode(new Raw("raw")); + + return new ValueTask(new WorkflowBuilder(start) + .AddEdge(start, raw) + .WithOutputFrom(raw) + .Build()); + } + + private sealed class Start : HostExecutor + { + public Start(string id) : base(id) { } + + protected override ValueTask ExecuteCoreAsync( + SampleContext input, IWorkflowContext context, CancellationToken cancellationToken) + => ValueTask.FromResult(input); + } + + private sealed class Raw : Executor + { + public Raw(string id) : base(id) { } + + public override ValueTask HandleAsync( + SampleContext message, IWorkflowContext context, CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SampleResult(message.Value)); + } + } +} + +public class GatePolicyRulesTests +{ + [Fact] + public void An_unlocked_declaration_accepts_anything() + { + var declared = new ApprovalGate { Mode = ExecutionMode.RequireApproval, RequiredApprovers = 3 }; + + GatePolicyRules.Violations(declared, ApprovalGate.Autonomous).Should().BeEmpty(); + GatePolicyRules.Reconcile(declared, ApprovalGate.Autonomous).Mode.Should().Be(ExecutionMode.Autonomous); + } + + [Theory] + [InlineData(ExecutionMode.RequireApproval, ExecutionMode.Autonomous, true)] + [InlineData(ExecutionMode.Conditional, ExecutionMode.Autonomous, true)] + [InlineData(ExecutionMode.Autonomous, ExecutionMode.RequireApproval, false)] + [InlineData(ExecutionMode.Conditional, ExecutionMode.RequireApproval, false)] + [InlineData(ExecutionMode.RequireApproval, ExecutionMode.RequireApproval, false)] + public void Mode_changes_against_a_locked_declaration(ExecutionMode declared, ExecutionMode candidate, bool loosens) + { + IReadOnlyList violations = GatePolicyRules.Violations( + new ApprovalGate { Mode = declared, Locked = true }, new ApprovalGate { Mode = candidate }); + + violations.Any().Should().Be(loosens); + } + + [Fact] + public void Every_weakened_field_of_a_locked_gate_is_reported() + { + var declared = new ApprovalGate + { + Mode = ExecutionMode.RequireApproval, + RequiredApprovers = 2, + AllowModification = false, + RequireSegregationOfDuties = true, + OnExpiry = ExpiryAction.DeadStop, + Locked = true + }; + + var candidate = new ApprovalGate + { + Mode = ExecutionMode.Autonomous, + RequiredApprovers = 1, + AllowModification = true, + RequireSegregationOfDuties = false, + OnExpiry = ExpiryAction.AutoApprove + }; + + GatePolicyRules.Violations(declared, candidate).Should().HaveCount(5); + } + + [Fact] + public void Reconcile_pulls_every_weakened_field_back_to_the_locked_declaration() + { + var declared = new ApprovalGate + { + Mode = ExecutionMode.RequireApproval, + RequiredApprovers = 2, + RequireSegregationOfDuties = true, + OnExpiry = ExpiryAction.DeadStop, + Locked = true + }; + + ApprovalGate effective = GatePolicyRules.Reconcile(declared, new ApprovalGate + { + Mode = ExecutionMode.Autonomous, + Reason = "tenant reason", + RequiredApprovers = 1, + AllowModification = true, + RequireSegregationOfDuties = false, + OnExpiry = ExpiryAction.AutoApprove + }); + + effective.Mode.Should().Be(ExecutionMode.RequireApproval); + effective.RequiredApprovers.Should().Be(2); + effective.AllowModification.Should().BeFalse(); + effective.RequireSegregationOfDuties.Should().BeTrue(); + effective.OnExpiry.Should().Be(ExpiryAction.DeadStop); + effective.Reason.Should().Be("tenant reason", "fields that do not weaken the gate still belong to the tenant"); + } + + [Fact] + public void Reconcile_keeps_a_tightening_change_on_a_locked_gate() + { + var declared = new ApprovalGate { Mode = ExecutionMode.RequireApproval, RequiredApprovers = 1, Locked = true }; + + ApprovalGate effective = GatePolicyRules.Reconcile( + declared, new ApprovalGate { Mode = ExecutionMode.RequireApproval, RequiredApprovers = 4 }); + + effective.RequiredApprovers.Should().Be(4); + } +} + +public class GateConfigurationServiceTests +{ + private const string Workflow = "configurable"; + private const string Version = "1.0.0"; + + [Fact] + public async Task Nodes_default_to_autonomous_with_no_tenant_configuration() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.GetNodesAsync(Workflow, Version, "t1", default); + + result.Kind.Should().Be(GateConfigResultKind.Ok); + ExecutorNodeDto validate = Node(result, "validate"); + validate.Effective.Mode.Should().Be("autonomous"); + validate.EffectiveSource.Should().Be(GateSources.Definition); + validate.TenantOverride.Should().BeNull(); + result.Nodes!.Nodes.Should().OnlyContain(n => n.TenantOverride == null); + } + + [Fact] + public async Task Unknown_workflow_and_version_are_not_found() + { + (IGateConfigurationService service, _) = Build(); + + (await service.GetNodesAsync("nope", Version, "t1", default)) + .Kind.Should().Be(GateConfigResultKind.UnknownWorkflow); + (await service.GetNodesAsync(Workflow, "9.9.9", "t1", default)) + .Kind.Should().Be(GateConfigResultKind.UnknownWorkflow); + } + + [Fact] + public async Task Version_latest_resolves_the_highest_registered_version() + { + var registry = new WorkflowRegistry([new ConfigurableWorkflow(), new ConfigurableWorkflow(version: "2.0.0")]); + var service = new GateConfigurationService(registry, new WorkflowInspector(), new InMemoryGatePolicyStore()); + + (await service.GetNodesAsync(Workflow, "latest", "t1", default)).Nodes!.WorkflowVersion.Should().Be("2.0.0"); + } + + [Fact] + public async Task Configuring_a_node_makes_it_require_approval_for_that_tenant_only() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult set = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("validate", new ExecutionPolicyDto("requireApproval", Reason: "tenant policy")), null, default); + + set.Kind.Should().Be(GateConfigResultKind.Ok); + ExecutorNodeDto configured = Node(set, "validate"); + configured.Effective.Mode.Should().Be("requireApproval"); + configured.Effective.Reason.Should().Be("tenant policy"); + configured.EffectiveSource.Should().Be(GateSources.Tenant); + configured.Declared.Mode.Should().Be("autonomous", "the declaration is unchanged"); + + GateConfigResult other = await service.GetNodesAsync(Workflow, Version, "t2", default); + Node(other, "validate").Effective.Mode.Should().Be("autonomous"); + Node(other, "validate").EffectiveSource.Should().Be(GateSources.Definition); + } + + [Fact] + public async Task A_tenant_may_turn_off_an_unlocked_declared_gate() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult set = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("notify", new ExecutionPolicyDto("autonomous")), null, default); + + set.Kind.Should().Be(GateConfigResultKind.Ok); + Node(set, "notify").Effective.Mode.Should().Be("autonomous"); + } + + [Fact] + public async Task Unspecified_fields_inherit_the_declared_gate() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult set = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("notify", new ExecutionPolicyDto("requireApproval", RequiredApprovers: 3)), null, default); + + ExecutionPolicyDto effective = Node(set, "notify").Effective; + effective.RequiredApprovers.Should().Be(3); + effective.Assignees.Should().ContainSingle() + .Which.Should().Be("group:ops", "the tenant did not restate the assignees"); + effective.ExpirySeconds.Should().Be((int)TimeSpan.FromHours(4).TotalSeconds); + effective.Reason.Should().Be("NotifiesCustomers"); + } + + [Fact] + public async Task Loosening_a_locked_node_is_rejected_and_persists_nothing() + { + (IGateConfigurationService service, InMemoryGatePolicyStore policies) = Build(); + + GateConfigResult result = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("pay", new ExecutionPolicyDto("autonomous")), null, default); + + result.Kind.Should().Be(GateConfigResultKind.Rejected); + result.Detail.Should().Contain("locked"); + (await policies.ListAsync("t1", Workflow, Version, default)).Should().BeEmpty(); + } + + [Fact] + public async Task Tightening_a_locked_node_is_allowed() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("pay", new ExecutionPolicyDto("requireApproval", RequiredApprovers: 3)), null, default); + + result.Kind.Should().Be(GateConfigResultKind.Ok); + ExecutionPolicyDto effective = Node(result, "pay").Effective; + effective.Mode.Should().Be("requireApproval"); + effective.RequiredApprovers.Should().Be(3); + effective.RequireSegregationOfDuties.Should().BeTrue("the locked declaration required it"); + } + + [Fact] + public async Task A_locked_node_is_flagged_so_a_ui_can_disable_the_control() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.GetNodesAsync(Workflow, Version, "t1", default); + + Node(result, "pay").Locked.Should().BeTrue(); + Node(result, "notify").Locked.Should().BeFalse(); + } + + [Fact] + public async Task An_unknown_executor_is_reported_without_touching_the_valid_ones() + { + (IGateConfigurationService service, InMemoryGatePolicyStore policies) = Build(); + + GateConfigResult result = await service.SetNodesAsync(Workflow, Version, "t1", new Dictionary + { + ["validate"] = new("requireApproval"), + ["ghost"] = new("requireApproval") + }, null, default); + + result.Kind.Should().Be(GateConfigResultKind.UnknownExecutor); + (await policies.ListAsync("t1", Workflow, Version, default)) + .Should().BeEmpty("a bulk write is all-or-nothing"); + } + + [Fact] + public async Task A_raw_node_cannot_be_configured() + { + var registry = new WorkflowRegistry([new RawOnlyWorkflow()]); + var service = new GateConfigurationService(registry, new WorkflowInspector(), new InMemoryGatePolicyStore()); + + GateConfigResult result = await service.SetNodesAsync("raw-only", "1.0.0", "t1", + Policy("raw", new ExecutionPolicyDto("requireApproval")), null, default); + + result.Kind.Should().Be(GateConfigResultKind.NotConfigurable); + } + + [Theory] + [InlineData("conditional")] + [InlineData("")] + [InlineData("whatever")] + public async Task Only_autonomous_and_require_approval_are_accepted(string mode) + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("validate", new ExecutionPolicyDto(mode)), null, default); + + result.Kind.Should().Be(GateConfigResultKind.Invalid); + result.Errors!["validate"].Should().ContainMatch("*autonomous*"); + } + + [Fact] + public async Task Out_of_range_values_are_reported_together() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("validate", new ExecutionPolicyDto( + "requireApproval", RequiredApprovers: 0, ExpirySeconds: 0, OnExpiry: "sometime")), + null, default); + + result.Kind.Should().Be(GateConfigResultKind.Invalid); + result.Errors!["validate"].Should().HaveCount(3); + } + + [Fact] + public async Task An_empty_request_is_rejected() + { + (IGateConfigurationService service, _) = Build(); + + GateConfigResult result = await service.SetNodesAsync( + Workflow, Version, "t1", new Dictionary(), null, default); + + result.Kind.Should().Be(GateConfigResultKind.Invalid); + result.Errors.Should().ContainKey("nodes"); + } + + [Fact] + public async Task Reset_returns_the_node_to_its_declared_gate() + { + (IGateConfigurationService service, _) = Build(); + + await service.SetNodesAsync(Workflow, Version, "t1", + Policy("notify", new ExecutionPolicyDto("autonomous")), null, default); + + GateConfigResult reset = await service.ResetNodeAsync(Workflow, Version, "t1", "notify", null, default); + + reset.Kind.Should().Be(GateConfigResultKind.Ok); + ExecutorNodeDto notify = Node(reset, "notify"); + notify.TenantOverride.Should().BeNull(); + notify.Effective.Mode.Should().Be("requireApproval"); + notify.EffectiveSource.Should().Be(GateSources.Definition); + } + + [Fact] + public async Task Reset_of_an_unknown_executor_is_reported() + { + (IGateConfigurationService service, _) = Build(); + + (await service.ResetNodeAsync(Workflow, Version, "t1", "ghost", null, default)) + .Kind.Should().Be(GateConfigResultKind.UnknownExecutor); + } + + [Fact] + public async Task A_host_wide_policy_applies_until_the_tenant_sets_its_own() + { + (IGateConfigurationService service, InMemoryGatePolicyStore policies) = Build(); + + await policies.SetAsync(null, Workflow, Version, "validate", + new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "host wide" }, default); + + ExecutorNodeDto beforeTenantConfig = Node(await service.GetNodesAsync(Workflow, Version, "t1", default), "validate"); + beforeTenantConfig.Effective.Mode.Should().Be("requireApproval"); + beforeTenantConfig.EffectiveSource.Should().Be(GateSources.Host); + beforeTenantConfig.TenantOverride.Should().BeNull(); + + GateConfigResult set = await service.SetNodesAsync(Workflow, Version, "t1", + Policy("validate", new ExecutionPolicyDto("autonomous")), null, default); + + Node(set, "validate").Effective.Mode.Should().Be("autonomous"); + Node(set, "validate").EffectiveSource.Should().Be(GateSources.Tenant); + } + + [Fact] + public async Task Configuration_is_scoped_to_one_workflow_version() + { + var registry = new WorkflowRegistry([new ConfigurableWorkflow(), new ConfigurableWorkflow(version: "2.0.0")]); + var service = new GateConfigurationService(registry, new WorkflowInspector(), new InMemoryGatePolicyStore()); + + await service.SetNodesAsync(Workflow, "1.0.0", "t1", + Policy("validate", new ExecutionPolicyDto("requireApproval")), null, default); + + Node(await service.GetNodesAsync(Workflow, "2.0.0", "t1", default), "validate") + .Effective.Mode.Should().Be("autonomous", "a policy belongs to the version it was written against"); + } + + [Fact] + public async Task Writes_are_audited() + { + var audit = new InMemoryAuditStore(); + var registry = new WorkflowRegistry([new ConfigurableWorkflow()]); + var service = new GateConfigurationService( + registry, new WorkflowInspector(), new InMemoryGatePolicyStore(), audit); + + await service.SetNodesAsync(Workflow, Version, "t1", + Policy("validate", new ExecutionPolicyDto("requireApproval")), null, default); + await service.ResetNodeAsync(Workflow, Version, "t1", "validate", null, default); + + IReadOnlyList entries = await audit.QueryAsync(null, 10, default); + entries.Select(e => e.Action).Should().Contain(["gate.policy.set", "gate.policy.reset"]); + entries.Should().OnlyContain(e => e.Detail!.Contains("\"tenantId\":\"t1\"")); + } + + private static (IGateConfigurationService Service, InMemoryGatePolicyStore Policies) Build() + { + var policies = new InMemoryGatePolicyStore(); + var registry = new WorkflowRegistry([new ConfigurableWorkflow()]); + return (new GateConfigurationService(registry, new WorkflowInspector(), policies), policies); + } + + private static Dictionary Policy(string executorId, ExecutionPolicyDto policy) + => new() { [executorId] = policy }; + + private static ExecutorNodeDto Node(GateConfigResult result, string executorId) + => result.Nodes!.Nodes.Single(n => n.ExecutorId == executorId); + + private sealed class RawOnlyWorkflow : IWorkflowDefinition + { + public string Name => "raw-only"; + public string Version => "1.0.0"; + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + ExecutorBinding raw = context.RawNode(new Raw("raw")); + return new ValueTask(new WorkflowBuilder(raw).WithOutputFrom(raw).Build()); + } + + private sealed class Raw : Executor + { + public Raw(string id) : base(id) { } + + public override ValueTask HandleAsync( + SampleContext message, IWorkflowContext context, CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SampleResult(message.Value)); + } + } +} diff --git a/tests/Abacus.Run.UnitTests/GateEvaluatorTests.cs b/tests/Abacus.Run.UnitTests/GateEvaluatorTests.cs index fe48442..d55e2c8 100644 --- a/tests/Abacus.Run.UnitTests/GateEvaluatorTests.cs +++ b/tests/Abacus.Run.UnitTests/GateEvaluatorTests.cs @@ -11,8 +11,9 @@ public class GateEvaluatorTests private static GateEvaluator Build( ApprovalGate? definitionGate = null, IGatePolicyStore? policies = null, - IApprovalStore? approvals = null) - => new("wf", "1.0.0", + IApprovalStore? approvals = null, + string? tenantId = "tenant-a") + => new("wf", "1.0.0", tenantId, definitionGate is null ? new Dictionary() : new Dictionary { ["pay"] = definitionGate }, @@ -81,7 +82,7 @@ public async Task Conditional_without_a_predicate_proceeds() public async Task Policy_store_overrides_the_definition_gate() { var policies = new InMemoryGatePolicyStore(); - await policies.SetAsync("wf", "1.0.0", "pay", + await policies.SetAsync(null, "wf", "1.0.0", "pay", new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "promoted at runtime" }, default); // An executor the definition left autonomous becomes gated with no redeploy. @@ -96,7 +97,7 @@ await policies.SetAsync("wf", "1.0.0", "pay", public async Task Instance_override_outranks_the_workflow_policy() { var policies = new InMemoryGatePolicyStore(); - await policies.SetAsync("wf", "1.0.0", "pay", new ApprovalGate { Mode = ExecutionMode.RequireApproval }, default); + await policies.SetAsync(null, "wf", "1.0.0", "pay", new ApprovalGate { Mode = ExecutionMode.RequireApproval }, default); await policies.SetInstanceOverrideAsync("i1", "pay", ApprovalGate.Autonomous, default); (await Build(ApprovalGate.Autonomous, policies).EvaluateAsync("i1", "pay", new Payload(), default)) @@ -106,6 +107,87 @@ public async Task Instance_override_outranks_the_workflow_policy() .Kind.Should().Be(GateOutcomeKind.Pause, "a different instance still sees the workflow policy"); } + [Fact] + public async Task Tenant_policy_outranks_the_host_wide_policy() + { + var policies = new InMemoryGatePolicyStore(); + await policies.SetAsync(null, "wf", "1.0.0", "pay", + new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "host default" }, default); + await policies.SetAsync("tenant-a", "wf", "1.0.0", "pay", ApprovalGate.Autonomous, default); + + (await Build(ApprovalGate.Autonomous, policies, tenantId: "tenant-a") + .EvaluateAsync("i1", "pay", new Payload(), default)) + .Kind.Should().Be(GateOutcomeKind.Proceed); + + (await Build(ApprovalGate.Autonomous, policies, tenantId: "tenant-b") + .EvaluateAsync("i2", "pay", new Payload(), default)) + .Kind.Should().Be(GateOutcomeKind.Pause, "a tenant without its own policy falls back to the host default"); + } + + [Fact] + public async Task One_tenants_policy_does_not_leak_into_another() + { + var policies = new InMemoryGatePolicyStore(); + await policies.SetAsync("tenant-a", "wf", "1.0.0", "pay", + new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "tenant a is cautious" }, default); + + (await Build(ApprovalGate.Autonomous, policies, tenantId: "tenant-a") + .EvaluateAsync("i1", "pay", new Payload(), default)) + .Kind.Should().Be(GateOutcomeKind.Pause); + + (await Build(ApprovalGate.Autonomous, policies, tenantId: "tenant-b") + .EvaluateAsync("i2", "pay", new Payload(), default)) + .Kind.Should().Be(GateOutcomeKind.Proceed, "tenant b never configured this executor"); + } + + [Fact] + public async Task Policy_cannot_un_gate_an_executor_the_definition_locked() + { + var declared = new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "sox", Locked = true }; + + var policies = new InMemoryGatePolicyStore(); + await policies.SetAsync("tenant-a", "wf", "1.0.0", "pay", ApprovalGate.Autonomous, default); + + (await Build(declared, policies, tenantId: "tenant-a").EvaluateAsync("i1", "pay", new Payload(), default)) + .Kind.Should().Be(GateOutcomeKind.Pause, "a locked gate is the author's floor"); + } + + [Fact] + public async Task Policy_may_still_tighten_a_locked_gate() + { + var declared = new ApprovalGate { Mode = ExecutionMode.Autonomous, Locked = true }; + + var policies = new InMemoryGatePolicyStore(); + await policies.SetAsync("tenant-a", "wf", "1.0.0", "pay", + new ApprovalGate { Mode = ExecutionMode.RequireApproval, Reason = "tenant wants eyes on this" }, default); + + GateOutcome outcome = await Build(declared, policies, tenantId: "tenant-a") + .EvaluateAsync("i1", "pay", new Payload(), default); + + outcome.Kind.Should().Be(GateOutcomeKind.Pause); + outcome.Gate!.Reason.Should().Be("tenant wants eyes on this"); + } + + [Fact] + public async Task Locked_conditional_gate_keeps_its_predicate_when_a_policy_tries_to_un_gate_it() + { + ApprovalGate declared = new ApprovalGateBuilder() + .When(p => p.Amount > 25_000m) + .Locked() + .Build(); + + var policies = new InMemoryGatePolicyStore(); + await policies.SetAsync("tenant-a", "wf", "1.0.0", "pay", ApprovalGate.Autonomous, default); + + GateEvaluator evaluator = Build(declared, policies, tenantId: "tenant-a"); + + (await evaluator.EvaluateAsync("i1", "pay", new Payload(Amount: 100m), default)) + .Kind.Should().Be(GateOutcomeKind.Proceed); + + (await evaluator.EvaluateAsync("i2", "pay", new Payload(Amount: 50_000m), default)) + .Kind.Should().Be(GateOutcomeKind.Pause, "the predicate survives the attempted downgrade"); + } + [Fact] public async Task Policy_store_failure_falls_back_to_the_definition_not_to_autonomous() { @@ -249,13 +331,24 @@ private static async Task CreateApproval( private sealed class ThrowingPolicyStore : IGatePolicyStore { public ValueTask FindAsync( - string workflowName, string workflowVersion, string executorId, string? instanceId, CancellationToken cancellationToken) + string? tenantId, string workflowName, string workflowVersion, string executorId, string? instanceId, + CancellationToken cancellationToken) + => throw new InvalidOperationException("policy store unavailable"); + + public ValueTask> ListAsync( + string? tenantId, string workflowName, string workflowVersion, CancellationToken cancellationToken) => throw new InvalidOperationException("policy store unavailable"); public ValueTask SetAsync( - string workflowName, string workflowVersion, string executorId, ApprovalGate gate, CancellationToken cancellationToken) + string? tenantId, string workflowName, string workflowVersion, string executorId, ApprovalGate gate, + CancellationToken cancellationToken) => ValueTask.CompletedTask; + public ValueTask RemoveAsync( + string? tenantId, string workflowName, string workflowVersion, string executorId, + CancellationToken cancellationToken) + => ValueTask.FromResult(false); + public ValueTask SetInstanceOverrideAsync( string instanceId, string executorId, ApprovalGate gate, CancellationToken cancellationToken) => ValueTask.CompletedTask;