From f5771a285c06d7b125bf620716e13b955ca20f7d Mon Sep 17 00:00:00 2001 From: Ninja Date: Tue, 11 Aug 2026 00:41:27 +0100 Subject: [PATCH 1/3] feat: add workflow audit records at the framework level MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A workflow definition can now declare the shape of its own audit record via IAuditedWorkflowDefinition, and the runtime hands every node an IWorkflowAuditRecorder bound to that declaration. Storage stays workflow-agnostic — a root plus string-typed entries with JSON payloads — so a new workflow needs no schema change. Recording is best-effort by contract: an audit write must never fail the work it describes, so store failures are logged and swallowed. GET /workflows/{name}/instances/{id}/state returns an instance's lifecycle status alongside whatever record its workflow declared, projected back through the declared sections so a run in progress shows what is still outstanding as readily as what is done. ?section= narrows the response. The framework default is InMemoryAuditRecordStore. Abacus.Run.Service displaces it with an EF Core SQLite store under Abacus:AuditRecords, whose connection string is resolved from IOptions inside the context factory rather than read at wire time, so a test host's configuration override actually applies. README.md and docs/wiki.md document the declaration, the recorder's guarantees, the state route, the configuration section, and the custom-store extension point. --- README.md | 79 +++++++- docs/wiki.md | 172 +++++++++++++++- .../Abacus.Run.Service.csproj | 5 + .../Auditing/AuditRecordDbContext.cs | 104 ++++++++++ .../AuditRecordServiceCollectionExtensions.cs | 80 ++++++++ ...0809164853_InitialAuditRecords.Designer.cs | 129 ++++++++++++ .../20260809164853_InitialAuditRecords.cs | 83 ++++++++ .../AuditRecordDbContextModelSnapshot.cs | 126 ++++++++++++ .../Auditing/SqliteAuditRecordStore.cs | 125 ++++++++++++ src/Abacus.Run.Service/Program.cs | 5 + src/Abacus.Run/Abstractions/Auditing.cs | 157 +++++++++++++++ src/Abacus.Run/Abstractions/HostExecutor.cs | 7 + .../Abstractions/WorkflowDefinition.cs | 12 +- src/Abacus.Run/Api/Endpoints.cs | 103 ++++++++++ src/Abacus.Run/Api/HostBuilderExtensions.cs | 2 + src/Abacus.Run/Core/WorkflowAuditRecorder.cs | 160 +++++++++++++++ src/Abacus.Run/Core/WorkflowRunner.cs | 43 +++- .../Persistence/InMemoryAuditRecordStore.cs | 66 +++++++ .../HostFixture.cs | 71 +++++++ .../InstanceStateTests.cs | 113 +++++++++++ .../Core/WorkflowAuditRecorderTests.cs | 184 ++++++++++++++++++ 21 files changed, 1809 insertions(+), 17 deletions(-) create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordDbContext.cs create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordServiceCollectionExtensions.cs create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.Designer.cs create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.cs create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/AuditRecordDbContextModelSnapshot.cs create mode 100644 src/Abacus.Run.Service/Infrastructure/Auditing/SqliteAuditRecordStore.cs create mode 100644 src/Abacus.Run/Abstractions/Auditing.cs create mode 100644 src/Abacus.Run/Core/WorkflowAuditRecorder.cs create mode 100644 src/Abacus.Run/Persistence/InMemoryAuditRecordStore.cs create mode 100644 tests/Abacus.Run.IntegrationTests/InstanceStateTests.cs create mode 100644 tests/Abacus.Run.UnitTests/Core/WorkflowAuditRecorderTests.cs diff --git a/README.md b/README.md index 46816ad..552b42c 100644 --- a/README.md +++ b/README.md @@ -285,6 +285,15 @@ Policies are keyed by workflow **version**, since executor ids and gates change `GET /instances/{id}/events` is an SSE stream. Instance queries support status, workflow, correlation ID, limit, and offset filters. +### Instance state and audit records + +- `GET /workflows/{name}/instances/{id}/state` + +Returns the instance's lifecycle status alongside the audit record its own workflow declared. The +route is scoped by workflow name because the response shape comes from that workflow's declaration; a +mismatched name returns `404`. `?section=plan,output` narrows the response to named sections. See +[Audit records](#audit-records). + ### Approvals - `GET /approvals` @@ -292,6 +301,60 @@ Policies are keyed by workflow **version**, since executor ids and gates change - `GET /instances/{id}/approvals` - `POST /approvals/{approvalId}/decision` +## Audit records + +Events record what the runtime did. An audit record answers the separate question of why a run's +result is defensible — the plan a node formed, the input it worked from, the output it produced. That +is workflow-specific, so the framework supplies the hook and the storage, never the schema. + +A definition opts in by implementing `IAuditedWorkflowDefinition` and declaring a root kind plus the +sections that may hang off it: + +```csharp +public sealed class OrderWorkflow + : IWorkflowDefinition, IAuditedWorkflowDefinition +{ + public AuditRecordDefinition AuditRecord { get; } = new( + "order", + "One order, as processed.", + [ + new AuditSectionDefinition("submission", "What was submitted.", Multiple: false), + new AuditSectionDefinition("step", "One processing step."), + new AuditSectionDefinition("outcome", "How the run settled.", Multiple: false) + ]); +} +``` + +The runtime then hands every node a recorder bound to that declaration. Executors reach it through +`Runtime.Audit`; a definition wiring its own nodes reads `WorkflowBuildContext.Audit`. Both are +nullable, so a workflow that declares no record costs nothing: + +```csharp +if (Runtime.Audit is { } audit) +{ + await audit.OpenAsync(input.OrderId, attributes: null, cancellationToken); + await audit.RecordAsync("step", Id, new { accepted = true }, cancellationToken); + await audit.CloseAsync(AuditRecordStatus.Completed, cancellationToken); +} +``` + +Storage stays workflow-agnostic: `IAuditRecordStore` keeps a root row plus entries whose section kind +is a string and whose payload is opaque JSON, so a new workflow needs no schema change. Recording is +best-effort by contract — a store failure is logged and swallowed, because failing work merely +because its explanation could not be filed trades a correct result for a missing one. Undeclared +section kinds are dropped, and re-recording the same `(section, key)` replaces the entry so a retried +executor corrects its record rather than contradicting it. + +Read the record back through `GET /workflows/{name}/instances/{id}/state`. Every declared section +appears whether or not anything has been recorded into it, so a run in progress shows what is still +outstanding as readily as what is done. + +The framework default is `InMemoryAuditRecordStore`. `Abacus.Run.Service` displaces it with an EF +Core SQLite store via `AddSqliteAuditRecords(configuration)`, configured under +`Abacus:AuditRecords:ConnectionString`. + +Full walkthrough: [Workflow audit records](docs/wiki.md#workflow-audit-records). + ## Configuration Options are read from the `WorkflowHost` configuration section. For example: @@ -319,10 +382,14 @@ Options are read from the `WorkflowHost` configuration section. For example: } ``` -The default host uses in-memory instance, event, log, approval, checkpoint, blob, and audit stores. Treat this configuration as development-oriented until durable store implementations are supplied. +The default host uses in-memory instance, event, log, approval, checkpoint, blob, audit, and audit-record stores. Treat this configuration as development-oriented until durable store implementations are supplied. Set `Abacus:SqlServer:ConnectionString` to enable the EF Core SQL Server stores and `Abacus:Redis:ConnectionString` to enable Redis Streams and control messages. `AddAbacus` keeps the in-memory stores when these settings are absent. +`Abacus:AuditRecords:ConnectionString` points the SQLite audit-record store at its database file and +defaults to `Data Source=./data/abacus-audit.db`. The directory is created and the migrations applied +at startup. + ## Project Layout | Project | Responsibility | @@ -340,11 +407,11 @@ Folders inside each project: src/Abacus.Run/ src/Abacus.Run.Service/ Abstractions/ ControlPlane/ Razor Pages backing services Api/ Infrastructure/ SQL Server stores, Redis bus - Core/ Pages/ control-plane Razor Pages - Dispatch/ wwwroot/ control-plane CSS and JS - EventBus/ Program.cs - Executors/ AbacusServiceCollectionExtensions.cs - Middlewares/ + Core/ Auditing/ audit-record store and migrations + Dispatch/ Pages/ control-plane Razor Pages + EventBus/ wwwroot/ control-plane CSS and JS + Executors/ Program.cs + Middlewares/ AbacusServiceCollectionExtensions.cs Persistence/ ``` diff --git a/docs/wiki.md b/docs/wiki.md index 7831af9..ac88a42 100644 --- a/docs/wiki.md +++ b/docs/wiki.md @@ -12,6 +12,7 @@ This page is the repository-level technical wiki. It documents the implementatio - [Project structure](#project-structure) - [Getting started](#getting-started) - [Authoring a workflow](#authoring-a-workflow) +- [Workflow audit records](#workflow-audit-records) - [Registering workflows and middleware](#registering-workflows-and-middleware) - [Instance lifecycle](#instance-lifecycle) - [Retries and failure classification](#retries-and-failure-classification) @@ -43,6 +44,7 @@ This page is the repository-level technical wiki. It documents the implementatio | Events | Sequenced per-instance event store plus optional event bus and SSE relay | | Approvals | Durable approval contracts and in-memory coordinator/store, with decision and expiry handling | | Middleware | Workflow-level and host-executor-level pipelines | +| Audit records | A workflow declares the shape of its own audit record; the runtime hands every node a recorder and stores entries generically | | Library | `src/Abacus.Run` — headless framework: runtime, dispatch, executors, middleware, in-memory stores, HTTP API | | Service host | `src/Abacus.Run.Service` — control-plane UI, SQL Server stores, Redis event bus, startup wiring | | Container | Multi-stage .NET 9 image listening on port 8080 | @@ -58,6 +60,7 @@ The repository contains a working runtime, API host, control plane, built-in exe - `InMemoryApprovalStore` - `InMemoryGatePolicyStore` - `InMemoryAuditStore` +- `InMemoryAuditRecordStore` - `InMemoryBlobStore` - `OverflowCheckpointStore` over the blob abstraction - `InMemoryEventBus` @@ -151,11 +154,11 @@ Folders inside each project: src/Abacus.Run/ src/Abacus.Run.Service/ Abstractions/ ControlPlane/ Razor Pages backing services Api/ Infrastructure/ SQL Server stores, Redis bus - Core/ Pages/ control-plane Razor Pages - Dispatch/ wwwroot/ control-plane CSS and JS - EventBus/ Program.cs - Executors/ AbacusServiceCollectionExtensions.cs - Middlewares/ + Core/ Auditing/ audit-record store and migrations + Dispatch/ Pages/ control-plane Razor Pages + EventBus/ wwwroot/ control-plane CSS and JS + Executors/ Program.cs + Middlewares/ AbacusServiceCollectionExtensions.cs Persistence/ ``` @@ -297,6 +300,86 @@ The output type must be a reference type because a gated executor returns `null` `WorkflowBuildContext.RawNode(...)` is an escape hatch for raw framework executor bindings and agent bindings. Raw nodes participate in the graph but do not receive host executor middleware and cannot be approval-gated. Use `Node(...)` with a `HostExecutor` when middleware or approvals are required. +## Workflow audit records + +Events answer "what did the runtime do". An audit record answers "why is this result defensible" — +the plan a node formed, the input it worked from, the output it produced, and what it published. +Those are workflow-specific questions, so the framework supplies the hook and the storage but not the +schema. + +The mechanism has three stages, and the separation between them is the point. + +**1. The definition declares the shape.** A definition that keeps a record implements +`IAuditedWorkflowDefinition` and returns an `AuditRecordDefinition`: a root aggregate kind plus the +child sections that may hang off it. + +```csharp +public sealed class ExampleWorkflowDefinition + : IWorkflowDefinition, IAuditedWorkflowDefinition +{ + public AuditRecordDefinition AuditRecord => ExampleAuditRecord.Definition; +} + +public static class ExampleAuditRecord +{ + public const string RootKind = "example-workflow"; + + // Section kinds are written into storage, so they are part of the workflow's contract — + // name them as constants rather than repeating string literals at each call site. + public const string Submission = "submission"; + public const string Plan = "plan"; + public const string Output = "output"; + + public static readonly AuditRecordDefinition Definition = new( + RootKind, + "A workflow-specific audit record for the important steps in a run.", + [ + new AuditSectionDefinition(Submission, "The input values and context used at start.", Multiple: false), + new AuditSectionDefinition(Plan, "The plan the workflow formed before acting."), + new AuditSectionDefinition(Output, "The resulting decision or artifact.") + ]); +} +``` + +**2. The runtime hands every node a recorder.** `WorkflowRunner` reads the interface at build time +and, when an `IAuditRecordStore` is registered, constructs a `WorkflowAuditRecorder` bound to the +definition and the instance. Executors reach it through `HostExecutorRuntime.Audit`; a definition +wiring its own nodes reads `WorkflowBuildContext.Audit`. Both are nullable — a workflow that declares +no record gets none, and no executor needs to know which is the case. + +```csharp +if (Runtime.Audit is { } audit) +{ + await audit.OpenAsync(input.RunId, attributes: null, cancellationToken); + await audit.RecordAsync(ExampleAuditRecord.Output, item.Id, result, cancellationToken); + await audit.CloseAsync(AuditRecordStatus.Completed, cancellationToken); +} +``` + +**3. Storage stays generic.** `IAuditRecordStore` holds a root row and a stream of entries whose +section kind is a string and whose payload is opaque JSON. A new workflow with a completely different +record needs no schema change. The framework default is `InMemoryAuditRecordStore`; +`Abacus.Run.Service` displaces it with a SQLite-backed store under `Abacus:AuditRecords`. + +Guarantees the recorder makes: + +| Rule | Reason | +| --- | --- | +| An undeclared section kind is logged and dropped | The declared shape is the contract, not a suggestion | +| Re-recording the same `(section, key)` replaces the entry | A retried executor corrects its record rather than contradicting it | +| Entries carry a monotonic sequence | Order of work survives storage that does not preserve insertion order | +| A store failure is swallowed and logged, never rethrown | An audit write explains work that already happened; failing the work because its explanation could not be filed trades a correct result for a missing one | +| `CloseAsync` before any `OpenAsync` writes nothing | A root with no identity is worse than no root | + +Because a failed write is invisible to the workflow, order matters at the edges: record a publication +failure *before* letting it propagate, so the record explains the failure it caused. + +The record is readable through the framework's own route — see +[Instances and diagnostics](#instances-and-diagnostics) — which projects the stored entries back +through the declared sections. Every declared section appears whether or not anything has been +recorded into it yet, so a caller reading a run in progress sees what is still outstanding as readily +as what is done. + ## Registering workflows and middleware The API host uses a fluent registration builder: @@ -619,6 +702,41 @@ Unknown workflow, version, or executor returns `404`. A raw node or an unsupport | `GET` | `/instances/{id}/checkpoints` | Inspect checkpoint metadata | | `GET` | `/instances/{id}/events/history` | Read persisted event history | | `GET` | `/instances/{id}/events` | Subscribe to live SSE events | +| `GET` | `/workflows/{name}/instances/{id}/state` | Lifecycle status plus the workflow's own audit record | + +`/workflows/{name}/instances/{id}/state` is the one instance route scoped by workflow name, because +what it returns is shaped by that workflow's declaration. A mismatched name is a wrong URL rather +than a different resource, so it returns `404` rather than the instance. `audit` is `null` when the +workflow declares no record. `?section=` narrows the response to named sections: + +```bash +curl 'http://localhost:5000/workflows/example-workflow/instances/{id}/state?section=plan,output' +``` + +```json +{ + "instance": { "instanceId": "…", "status": "Completed", "workflowVersion": "1.0.0" }, + "audit": { + "rootKind": "example-workflow", + "rootKey": "RUN-4471", + "status": "Completed", + "attributes": { "requestId": "…" }, + "sections": [ + { + "kind": "output", + "description": "The resulting decision or artifact.", + "multiple": true, + "entries": [ + { "key": "OUT-001", "sequence": 12, "recordedUtc": "…", "payload": { "decision": "Approved" } } + ] + } + ] + } +} +``` + +Payloads are re-emitted as JSON rather than as escaped strings, and sections the workflow has not yet +recorded into come back with an empty `entries` array rather than being omitted. ### Instance controls @@ -711,6 +829,28 @@ Configuration is bound from the `WorkflowHost` section. Defaults are defined in } ``` +### Host-supplied sections + +`WorkflowHost` is the framework's own section. `Abacus.Run.Service` reads further sections that +select the concrete infrastructure it substitutes for the in-memory defaults: + +```json +{ + "Abacus": { + "AuditRecords": { "ConnectionString": "Data Source=./data/abacus-audit.db" }, + "SqlServer": { "ConnectionString": "", "EnsureDatabaseCreated": false }, + "Redis": { "ConnectionString": "", "MaxStreamLength": 10000 } + } +} +``` + +`Abacus:AuditRecords:ConnectionString` backs the generic audit-record store with SQLite and defaults +to `Data Source=./data/abacus-audit.db`; the directory is created at startup and the migrations are +applied by a hosted service. The connection string is resolved from `IOptions` inside the context +factory rather than read at registration time, so a test host's configuration override actually +applies — reading it at wire time silently binds every host to the deployed database file. Remove the +`AddSqliteAuditRecords` call to keep the framework's in-memory default. + ### Important production settings - Set a stable `ReplicaId` when running multiple hosts. @@ -874,8 +1014,8 @@ The solution includes several test layers: | Suite | Purpose | | --- | --- | -| Unit tests | Contracts, policies, runner behavior, middleware, executors, stores, and control logic | -| Integration tests | Real ASP.NET Core host, HTTP routes, event streams, approvals, and instance controls | +| Unit tests | Contracts, policies, runner behavior, middleware, executors, stores, audit recorder, and control logic | +| Integration tests | Real ASP.NET Core host, HTTP routes, event streams, approvals, instance controls, and the instance state route | | Chaos tests | Failure and lifecycle scenarios | | Load tests | Throughput-oriented test project | @@ -907,7 +1047,7 @@ dotnet test --configuration Release --no-build ### Custom persistence -Implement the store interfaces used by `AddWorkflowHost`, including instance, event, log, approval, gate policy, audit, blob, and checkpoint contracts. Preserve these invariants: +Implement the store interfaces used by `AddWorkflowHost`, including instance, event, log, approval, gate policy, audit, audit record, blob, and checkpoint contracts. Preserve these invariants: - Instance updates use optimistic concurrency. - Terminal state is not overwritten by a losing writer. @@ -915,6 +1055,15 @@ Implement the store interfaces used by `AddWorkflowHost`, including instance, ev - Event sequence values are unique and ordered per instance. - Checkpoint indexes are returned in the order expected by `CheckpointManager`. - Approval decisions are single-winner and quorum-aware. +- Audit record entries are unique per `(instance, section kind, key)`, so a re-record replaces. + +### Custom audit record storage + +Implement `IAuditRecordStore` when records must outlive the process or be queried outside the host. +Keep the payload opaque — the value of the contract is that a new workflow with a different record +shape needs no schema change. `Abacus.Run.Service/Infrastructure/Auditing` is a worked example: an EF +Core SQLite store with a unique index on `(InstanceId, SectionKind, Key)`, registered through +`AddSqliteAuditRecords()`, which removes the framework's in-memory registration rather than racing it. ### Custom executors @@ -994,6 +1143,13 @@ A tenant or host-wide policy is gating it. Call `GET /workflows/{name}/versions/ ### 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. +### `audit` is null on the instance state route + +The workflow does not implement `IAuditedWorkflowDefinition`, so it declares no record. If it does +implement it and the record is still empty, check that an `IAuditRecordStore` is registered — the +runner logs a warning and returns no recorder when a definition declares a record with no store +behind it — and remember that recorder failures are swallowed by design, so the host log is where a +storage problem surfaces, not the response. ### An outbound call is blocked diff --git a/src/Abacus.Run.Service/Abacus.Run.Service.csproj b/src/Abacus.Run.Service/Abacus.Run.Service.csproj index 8ce143a..cca9d11 100644 --- a/src/Abacus.Run.Service/Abacus.Run.Service.csproj +++ b/src/Abacus.Run.Service/Abacus.Run.Service.csproj @@ -7,6 +7,11 @@ + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordDbContext.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordDbContext.cs new file mode 100644 index 0000000..c3ec448 --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordDbContext.cs @@ -0,0 +1,104 @@ +using System.ComponentModel.DataAnnotations; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +namespace Abacus.Run.Service.Infrastructure.Auditing; + +/// +/// Root aggregate row. Deliberately workflow-agnostic: RootKind and RootKey carry +/// whatever the workflow declared, so a new workflow needs no schema change. +/// +public sealed class AuditRecordRootRow +{ + [Key, MaxLength(128)] + public string InstanceId { get; set; } = default!; + + [Required, MaxLength(128)] + public string WorkflowName { get; set; } = default!; + + [Required, MaxLength(32)] + public string WorkflowVersion { get; set; } = default!; + + [Required, MaxLength(64)] + public string RootKind { get; set; } = default!; + + [Required, MaxLength(256)] + public string RootKey { get; set; } = default!; + + [Required, MaxLength(32)] + public string Status { get; set; } = default!; + + public string AttributesJson { get; set; } = "{}"; + + public DateTimeOffset OpenedUtc { get; set; } + public DateTimeOffset? ClosedUtc { get; set; } + + public ICollection Entries { get; set; } = []; +} + +/// One child construct. The payload is opaque JSON — its shape is the workflow's business. +public sealed class AuditRecordEntryRow +{ + [Key] + public Guid Id { get; set; } + + [Required, MaxLength(128)] + public string InstanceId { get; set; } = default!; + public AuditRecordRootRow Root { get; set; } = default!; + + [Required, MaxLength(64)] + public string SectionKind { get; set; } = default!; + + [MaxLength(128)] + public string? Key { get; set; } + + public string PayloadJson { get; set; } = "null"; + + public int Sequence { get; set; } + + public DateTimeOffset RecordedUtc { get; set; } +} + +/// +/// Generic storage for workflow audit records. Owned by the host rather than the framework: the +/// framework defines the contract, a deployment chooses what it is written to. +/// +public sealed class AuditRecordDbContext(DbContextOptions options) : DbContext(options) +{ + public DbSet AuditRecords => Set(); + public DbSet AuditRecordEntries => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + // SQLite has no native DateTimeOffset — ORDER BY on the default string mapping throws. + var dtoConverter = new DateTimeOffsetToBinaryConverter(); + foreach (var entity in modelBuilder.Model.GetEntityTypes()) + { + foreach (var property in entity.GetProperties()) + { + if (property.ClrType == typeof(DateTimeOffset) || property.ClrType == typeof(DateTimeOffset?)) + { + property.SetValueConverter(dtoConverter); + } + } + } + + modelBuilder.Entity(b => + { + b.HasIndex(x => new { x.WorkflowName, x.RootKey }); + b.HasIndex(x => new { x.WorkflowName, x.Status }); + }); + + modelBuilder.Entity(b => + { + b.HasOne(x => x.Root) + .WithMany(x => x.Entries) + .HasForeignKey(x => x.InstanceId) + .OnDelete(DeleteBehavior.Cascade); + + // One entry per (instance, kind, key): a retried executor corrects its record instead of + // appending a second, contradictory one. + b.HasIndex(x => new { x.InstanceId, x.SectionKind, x.Key }).IsUnique(); + }); + } +} diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordServiceCollectionExtensions.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordServiceCollectionExtensions.cs new file mode 100644 index 0000000..550e27f --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/AuditRecordServiceCollectionExtensions.cs @@ -0,0 +1,80 @@ +using Abacus.Run.Abstractions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; + +namespace Abacus.Run.Service.Infrastructure.Auditing; + +public sealed class AuditRecordOptions +{ + public const string SectionName = "Abacus:AuditRecords"; + + /// SQLite connection string for the generic audit-record store. + public string ConnectionString { get; set; } = "Data Source=./data/abacus-audit.db"; +} + +/// +/// Replaces the framework's in-memory audit-record store with durable SQLite storage. The store is +/// generic: every workflow that declares an audit record writes here, and the record's meaning stays +/// with the definition that declared it. +/// +public static class AuditRecordServiceCollectionExtensions +{ + public static IServiceCollection AddSqliteAuditRecords( + this IServiceCollection services, IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(configuration); + + services + .AddOptions() + .Bind(configuration.GetSection(AuditRecordOptions.SectionName)); + + services.AddDbContextFactory((sp, builder) => + { + string connectionString = sp.GetRequiredService>().Value.ConnectionString; + EnsureDataDirectoryExists(connectionString); + builder.UseSqlite(connectionString); + }); + + // Displaces the framework default registered by AddAbacus. + services.RemoveAll(); + services.AddSingleton(); + + services.AddHostedService(); + + return services; + } + + private static void EnsureDataDirectoryExists(string connectionString) + { + const string key = "Data Source="; + int i = connectionString.IndexOf(key, StringComparison.OrdinalIgnoreCase); + if (i < 0) return; + + string tail = connectionString[(i + key.Length)..]; + int end = tail.IndexOf(';'); + string path = (end < 0 ? tail : tail[..end]).Trim().Trim('"'); + + string? dir = Path.GetDirectoryName(path); + if (!string.IsNullOrWhiteSpace(dir) && !Directory.Exists(dir)) + { + Directory.CreateDirectory(dir); + } + } +} + +/// Applies the audit-record migrations at startup. +public sealed class AuditRecordDatabaseInitializer(IDbContextFactory factory) : IHostedService +{ + public async Task StartAsync(CancellationToken cancellationToken) + { + await using AuditRecordDbContext db = await factory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + await db.Database.MigrateAsync(cancellationToken).ConfigureAwait(false); + } + + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; +} diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.Designer.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.Designer.cs new file mode 100644 index 0000000..6440187 --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.Designer.cs @@ -0,0 +1,129 @@ +// +using System; +using Abacus.Run.Service.Infrastructure.Auditing; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +#nullable disable + +namespace Abacus.Run.Service.Infrastructure.Auditing.Migrations +{ + [DbContext(typeof(AuditRecordDbContext))] + [Migration("20260809164853_InitialAuditRecords")] + partial class InitialAuditRecords + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder.HasAnnotation("ProductVersion", "9.0.9"); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordEntryRow", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("TEXT"); + + b.Property("InstanceId") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("Key") + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("PayloadJson") + .IsRequired() + .HasColumnType("TEXT"); + + b.Property("RecordedUtc") + .HasColumnType("INTEGER"); + + b.Property("SectionKind") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("TEXT"); + + b.Property("Sequence") + .HasColumnType("INTEGER"); + + b.HasKey("Id"); + + b.HasIndex("InstanceId", "SectionKind", "Key") + .IsUnique(); + + b.ToTable("AuditRecordEntries"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", b => + { + b.Property("InstanceId") + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("AttributesJson") + .IsRequired() + .HasColumnType("TEXT"); + + b.Property("ClosedUtc") + .HasColumnType("INTEGER"); + + b.Property("OpenedUtc") + .HasColumnType("INTEGER"); + + b.Property("RootKey") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("TEXT"); + + b.Property("RootKind") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("TEXT"); + + b.Property("Status") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("TEXT"); + + b.Property("WorkflowName") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("WorkflowVersion") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("TEXT"); + + b.HasKey("InstanceId"); + + b.HasIndex("WorkflowName", "RootKey"); + + b.HasIndex("WorkflowName", "Status"); + + b.ToTable("AuditRecords"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordEntryRow", b => + { + b.HasOne("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", "Root") + .WithMany("Entries") + .HasForeignKey("InstanceId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Root"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", b => + { + b.Navigation("Entries"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.cs new file mode 100644 index 0000000..f8c22a1 --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/20260809164853_InitialAuditRecords.cs @@ -0,0 +1,83 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace Abacus.Run.Service.Infrastructure.Auditing.Migrations +{ + /// + public partial class InitialAuditRecords : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "AuditRecords", + columns: table => new + { + InstanceId = table.Column(type: "TEXT", maxLength: 128, nullable: false), + WorkflowName = table.Column(type: "TEXT", maxLength: 128, nullable: false), + WorkflowVersion = table.Column(type: "TEXT", maxLength: 32, nullable: false), + RootKind = table.Column(type: "TEXT", maxLength: 64, nullable: false), + RootKey = table.Column(type: "TEXT", maxLength: 256, nullable: false), + Status = table.Column(type: "TEXT", maxLength: 32, nullable: false), + AttributesJson = table.Column(type: "TEXT", nullable: false), + OpenedUtc = table.Column(type: "INTEGER", nullable: false), + ClosedUtc = table.Column(type: "INTEGER", nullable: true) + }, + constraints: table => + { + table.PrimaryKey("PK_AuditRecords", x => x.InstanceId); + }); + + migrationBuilder.CreateTable( + name: "AuditRecordEntries", + columns: table => new + { + Id = table.Column(type: "TEXT", nullable: false), + InstanceId = table.Column(type: "TEXT", maxLength: 128, nullable: false), + SectionKind = table.Column(type: "TEXT", maxLength: 64, nullable: false), + Key = table.Column(type: "TEXT", maxLength: 128, nullable: true), + PayloadJson = table.Column(type: "TEXT", nullable: false), + Sequence = table.Column(type: "INTEGER", nullable: false), + RecordedUtc = table.Column(type: "INTEGER", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_AuditRecordEntries", x => x.Id); + table.ForeignKey( + name: "FK_AuditRecordEntries_AuditRecords_InstanceId", + column: x => x.InstanceId, + principalTable: "AuditRecords", + principalColumn: "InstanceId", + onDelete: ReferentialAction.Cascade); + }); + + migrationBuilder.CreateIndex( + name: "IX_AuditRecordEntries_InstanceId_SectionKind_Key", + table: "AuditRecordEntries", + columns: new[] { "InstanceId", "SectionKind", "Key" }, + unique: true); + + migrationBuilder.CreateIndex( + name: "IX_AuditRecords_WorkflowName_RootKey", + table: "AuditRecords", + columns: new[] { "WorkflowName", "RootKey" }); + + migrationBuilder.CreateIndex( + name: "IX_AuditRecords_WorkflowName_Status", + table: "AuditRecords", + columns: new[] { "WorkflowName", "Status" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "AuditRecordEntries"); + + migrationBuilder.DropTable( + name: "AuditRecords"); + } + } +} diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/AuditRecordDbContextModelSnapshot.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/AuditRecordDbContextModelSnapshot.cs new file mode 100644 index 0000000..b48e343 --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/Migrations/AuditRecordDbContextModelSnapshot.cs @@ -0,0 +1,126 @@ +// +using System; +using Abacus.Run.Service.Infrastructure.Auditing; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +#nullable disable + +namespace Abacus.Run.Service.Infrastructure.Auditing.Migrations +{ + [DbContext(typeof(AuditRecordDbContext))] + partial class AuditRecordDbContextModelSnapshot : ModelSnapshot + { + protected override void BuildModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder.HasAnnotation("ProductVersion", "9.0.9"); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordEntryRow", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("TEXT"); + + b.Property("InstanceId") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("Key") + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("PayloadJson") + .IsRequired() + .HasColumnType("TEXT"); + + b.Property("RecordedUtc") + .HasColumnType("INTEGER"); + + b.Property("SectionKind") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("TEXT"); + + b.Property("Sequence") + .HasColumnType("INTEGER"); + + b.HasKey("Id"); + + b.HasIndex("InstanceId", "SectionKind", "Key") + .IsUnique(); + + b.ToTable("AuditRecordEntries"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", b => + { + b.Property("InstanceId") + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("AttributesJson") + .IsRequired() + .HasColumnType("TEXT"); + + b.Property("ClosedUtc") + .HasColumnType("INTEGER"); + + b.Property("OpenedUtc") + .HasColumnType("INTEGER"); + + b.Property("RootKey") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("TEXT"); + + b.Property("RootKind") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("TEXT"); + + b.Property("Status") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("TEXT"); + + b.Property("WorkflowName") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("TEXT"); + + b.Property("WorkflowVersion") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("TEXT"); + + b.HasKey("InstanceId"); + + b.HasIndex("WorkflowName", "RootKey"); + + b.HasIndex("WorkflowName", "Status"); + + b.ToTable("AuditRecords"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordEntryRow", b => + { + b.HasOne("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", "Root") + .WithMany("Entries") + .HasForeignKey("InstanceId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Root"); + }); + + modelBuilder.Entity("Abacus.Run.Service.Infrastructure.Auditing.AuditRecordRootRow", b => + { + b.Navigation("Entries"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Abacus.Run.Service/Infrastructure/Auditing/SqliteAuditRecordStore.cs b/src/Abacus.Run.Service/Infrastructure/Auditing/SqliteAuditRecordStore.cs new file mode 100644 index 0000000..c0a4e6e --- /dev/null +++ b/src/Abacus.Run.Service/Infrastructure/Auditing/SqliteAuditRecordStore.cs @@ -0,0 +1,125 @@ +using Abacus.Run.Abstractions; +using Microsoft.EntityFrameworkCore; + +namespace Abacus.Run.Service.Infrastructure.Auditing; + +/// +/// SQLite-backed . Storage is generic by design — the framework hands +/// it a root and a stream of JSON-payload entries, and the workflow that declared them is the only +/// thing that knows what they mean. +/// +public sealed class SqliteAuditRecordStore(IDbContextFactory dbFactory) : IAuditRecordStore +{ + public async ValueTask UpsertRootAsync(AuditRecordRoot root, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(root); + + await using AuditRecordDbContext db = await dbFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + AuditRecordRootRow? existing = await db.AuditRecords + .FirstOrDefaultAsync(r => r.InstanceId == root.InstanceId, cancellationToken).ConfigureAwait(false); + + AuditRecordRootRow row = existing ?? new AuditRecordRootRow { InstanceId = root.InstanceId }; + + row.WorkflowName = root.WorkflowName; + row.WorkflowVersion = root.WorkflowVersion; + row.RootKind = root.RootKind; + row.RootKey = root.RootKey; + row.Status = root.Status; + row.AttributesJson = root.AttributesJson; + row.OpenedUtc = root.OpenedUtc; + row.ClosedUtc = root.ClosedUtc; + + if (existing is null) db.AuditRecords.Add(row); + + await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + + public async ValueTask AppendAsync(AuditRecordEntry entry, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(entry); + + await using AuditRecordDbContext db = await dbFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + AuditRecordEntryRow? existing = await db.AuditRecordEntries + .FirstOrDefaultAsync( + e => e.InstanceId == entry.InstanceId && e.SectionKind == entry.SectionKind && e.Key == entry.Key, + cancellationToken).ConfigureAwait(false); + + AuditRecordEntryRow row = existing ?? new AuditRecordEntryRow + { + Id = entry.Id, + InstanceId = entry.InstanceId, + SectionKind = entry.SectionKind, + Key = entry.Key + }; + + row.PayloadJson = entry.PayloadJson; + row.Sequence = entry.Sequence; + row.RecordedUtc = entry.RecordedUtc; + + if (existing is null) db.AuditRecordEntries.Add(row); + + await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + + public async ValueTask GetAsync(string instanceId, CancellationToken cancellationToken) + { + await using AuditRecordDbContext db = await dbFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + AuditRecordRootRow? root = await db.AuditRecords + .AsNoTracking() + .Include(r => r.Entries) + .FirstOrDefaultAsync(r => r.InstanceId == instanceId, cancellationToken).ConfigureAwait(false); + + if (root is null) return null; + + return new AuditRecordDocument( + Map(root), + [.. root.Entries.OrderBy(e => e.Sequence).Select(Map)]); + } + + public async ValueTask> ListAsync( + string workflowName, string? rootKey, string? status, int limit, CancellationToken cancellationToken) + { + await using AuditRecordDbContext db = await dbFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + IQueryable query = db.AuditRecords + .AsNoTracking() + .Where(r => r.WorkflowName == workflowName); + + if (rootKey is not null) query = query.Where(r => r.RootKey == rootKey); + if (status is not null) query = query.Where(r => r.Status == status); + + List rows = await query + .OrderByDescending(r => r.OpenedUtc) + .Take(Math.Max(1, limit)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + return [.. rows.Select(Map)]; + } + + private static AuditRecordRoot Map(AuditRecordRootRow row) => new() + { + InstanceId = row.InstanceId, + WorkflowName = row.WorkflowName, + WorkflowVersion = row.WorkflowVersion, + RootKind = row.RootKind, + RootKey = row.RootKey, + Status = row.Status, + AttributesJson = row.AttributesJson, + OpenedUtc = row.OpenedUtc, + ClosedUtc = row.ClosedUtc + }; + + private static AuditRecordEntry Map(AuditRecordEntryRow row) => new() + { + Id = row.Id, + InstanceId = row.InstanceId, + SectionKind = row.SectionKind, + Key = row.Key, + PayloadJson = row.PayloadJson, + Sequence = row.Sequence, + RecordedUtc = row.RecordedUtc + }; +} diff --git a/src/Abacus.Run.Service/Program.cs b/src/Abacus.Run.Service/Program.cs index 9fb39d6..8604b8d 100644 --- a/src/Abacus.Run.Service/Program.cs +++ b/src/Abacus.Run.Service/Program.cs @@ -1,5 +1,6 @@ using Abacus.Run.Api; using Abacus.Run.Service.ControlPlane; +using Abacus.Run.Service.Infrastructure.Auditing; using Abacus.Run.Core; using Abacus.Run.Service; using Microsoft.AspNetCore.Builder; @@ -11,6 +12,10 @@ builder.Services.AddProblemDetails(); builder.Services.AddAbacus(builder.Configuration); +// Durable, workflow-agnostic storage for the audit records that workflow definitions declare. +// Displaces the framework's in-memory default. +builder.Services.AddSqliteAuditRecords(builder.Configuration); + WebApplication app = builder.Build(); app.UseExceptionHandler(); diff --git a/src/Abacus.Run/Abstractions/Auditing.cs b/src/Abacus.Run/Abstractions/Auditing.cs new file mode 100644 index 0000000..e3d139b --- /dev/null +++ b/src/Abacus.Run/Abstractions/Auditing.cs @@ -0,0 +1,157 @@ +namespace Abacus.Run.Abstractions; + +/// +/// One kind of child construct a workflow's audit record may contain — a retrieval plan, an +/// execution input, an output, whatever that workflow considers audit-significant. +/// +/// Stable identifier written into storage. Treat as part of the workflow's contract. +/// What an investigator will find here. +/// False when at most one entry of this kind belongs to a record. +public sealed record AuditSectionDefinition(string Kind, string Description, bool Multiple = true); + +/// +/// The shape of a workflow's audit record: a root aggregate plus the child constructs that may hang +/// off it. Declared by the workflow definition, because only the workflow knows what is worth +/// auditing about its own run — the framework supplies the hook and the storage, not the schema. +/// +public sealed class AuditRecordDefinition +{ + private readonly Dictionary _sections; + + public AuditRecordDefinition( + string rootKind, + string description, + IReadOnlyList sections) + { + ArgumentException.ThrowIfNullOrWhiteSpace(rootKind); + ArgumentNullException.ThrowIfNull(sections); + + RootKind = rootKind; + Description = description ?? string.Empty; + Sections = sections; + + _sections = sections.ToDictionary(s => s.Kind, StringComparer.OrdinalIgnoreCase); + } + + /// Kind of the root aggregate — the thing one run of the workflow is about. + public string RootKind { get; } + + public string Description { get; } + + public IReadOnlyList Sections { get; } + + public bool Allows(string sectionKind) => + !string.IsNullOrWhiteSpace(sectionKind) && _sections.ContainsKey(sectionKind); + + public AuditSectionDefinition? Section(string sectionKind) => + _sections.GetValueOrDefault(sectionKind); +} + +/// +/// Implemented by a workflow definition that keeps an audit record. The runtime reads this at build +/// time and hands the definition's executors a recorder bound to it. +/// +public interface IAuditedWorkflowDefinition +{ + AuditRecordDefinition AuditRecord { get; } +} + +/// +/// The hook a workflow's executors use to build up the audit record as the run progresses. Obtained +/// from inside an executor, or from +/// when the definition wires its own nodes. +/// +/// +/// Recording is best-effort by contract: an audit write must never fail the work it describes. +/// Implementations swallow and log storage failures rather than propagating them. +/// +public interface IWorkflowAuditRecorder +{ + /// The shape this recorder accepts, as declared by the workflow definition. + AuditRecordDefinition Definition { get; } + + /// + /// Opens (or re-opens, on a resumed instance) the root aggregate. is + /// the workflow's own identifier for the thing being audited — a case reference, an order id. + /// + ValueTask OpenAsync( + string rootKey, + IReadOnlyDictionary? attributes, + CancellationToken cancellationToken); + + /// + /// Appends one child construct. must be declared by the + /// definition; groups entries within a kind (a check id, a node id). + /// + ValueTask RecordAsync( + string sectionKind, + string? key, + object? payload, + CancellationToken cancellationToken); + + /// Settles the record. Called on the terminal path of the workflow. + ValueTask CloseAsync(string status, CancellationToken cancellationToken); +} + +/// Root aggregate row as the generic store holds it. +public sealed record AuditRecordRoot +{ + public required string InstanceId { get; init; } + public required string WorkflowName { get; init; } + public required string WorkflowVersion { get; init; } + public required string RootKind { get; init; } + public required string RootKey { get; init; } + public required string Status { get; init; } + public string AttributesJson { get; init; } = "{}"; + public DateTimeOffset OpenedUtc { get; init; } = DateTimeOffset.UtcNow; + public DateTimeOffset? ClosedUtc { get; init; } +} + +/// One child construct hanging off a root. Payload is opaque JSON to the framework. +public sealed record AuditRecordEntry +{ + public required Guid Id { get; init; } + public required string InstanceId { get; init; } + public required string SectionKind { get; init; } + public string? Key { get; init; } + public required string PayloadJson { get; init; } + public required int Sequence { get; init; } + public DateTimeOffset RecordedUtc { get; init; } = DateTimeOffset.UtcNow; +} + +/// A complete audit record: the root and everything recorded against it. +public sealed record AuditRecordDocument(AuditRecordRoot Root, IReadOnlyList Entries); + +/// +/// Generic backing storage for audit records. Deliberately workflow-agnostic — it stores a root, a +/// stream of typed-by-string entries, and JSON payloads, so a new workflow needs no schema change. +/// +public interface IAuditRecordStore +{ + /// Creates or updates the root. Keyed by instance id, so a resumed run reuses its record. + ValueTask UpsertRootAsync(AuditRecordRoot root, CancellationToken cancellationToken); + + /// + /// Appends an entry. Replaces any existing entry with the same (instance, kind, key) so a retried + /// executor corrects its record rather than appending a second, contradictory one. + /// + ValueTask AppendAsync(AuditRecordEntry entry, CancellationToken cancellationToken); + + ValueTask GetAsync(string instanceId, CancellationToken cancellationToken); + + /// Roots for one workflow, newest first, optionally filtered by root key or status. + ValueTask> ListAsync( + string workflowName, + string? rootKey, + string? status, + int limit, + CancellationToken cancellationToken); +} + +/// Statuses the framework writes; a workflow may use its own beyond these. +public static class AuditRecordStatus +{ + public const string Open = "Open"; + public const string Completed = "Completed"; + public const string Failed = "Failed"; +} diff --git a/src/Abacus.Run/Abstractions/HostExecutor.cs b/src/Abacus.Run/Abstractions/HostExecutor.cs index aa31fef..773db86 100644 --- a/src/Abacus.Run/Abstractions/HostExecutor.cs +++ b/src/Abacus.Run/Abstractions/HostExecutor.cs @@ -78,6 +78,13 @@ public sealed class HostExecutorRuntime public IApprovalCoordinator? Approvals { get; init; } public IServiceProvider? Services { get; init; } + /// + /// The audit hook for this instance, present when the workflow definition declares an audit + /// record. Executors call it to add their own constructs — plans, inputs, outputs — as the run + /// progresses. Null when the workflow keeps no audit record. + /// + public IWorkflowAuditRecorder? Audit { get; init; } + /// Called when a host executor begins handling a message, before gate evaluation. public Func? ExecutorInvoked { get; init; } diff --git a/src/Abacus.Run/Abstractions/WorkflowDefinition.cs b/src/Abacus.Run/Abstractions/WorkflowDefinition.cs index 35052c5..e2930c6 100644 --- a/src/Abacus.Run/Abstractions/WorkflowDefinition.cs +++ b/src/Abacus.Run/Abstractions/WorkflowDefinition.cs @@ -58,7 +58,8 @@ public WorkflowBuildContext( string workflowVersion, int attempt, IServiceProvider? services, - Func attach) + Func attach, + IWorkflowAuditRecorder? audit = null) { InstanceId = instanceId; TenantId = tenantId; @@ -66,6 +67,7 @@ public WorkflowBuildContext( WorkflowVersion = workflowVersion; Attempt = attempt; Services = services; + Audit = audit; _attach = attach ?? throw new ArgumentNullException(nameof(attach)); } @@ -76,6 +78,14 @@ public WorkflowBuildContext( public int Attempt { get; } public IServiceProvider? Services { get; } + /// + /// The audit hook for this instance, present when the definition implements + /// . Available here so a definition can hand it to the + /// executors it constructs; executors attached with also reach it through + /// . + /// + public IWorkflowAuditRecorder? Audit { get; } + /// Gates declared during this build, by executor id. Read by the runtime. public IReadOnlyDictionary Gates => _gates; diff --git a/src/Abacus.Run/Api/Endpoints.cs b/src/Abacus.Run/Api/Endpoints.cs index e221845..6e94b1e 100644 --- a/src/Abacus.Run/Api/Endpoints.cs +++ b/src/Abacus.Run/Api/Endpoints.cs @@ -227,6 +227,45 @@ private static void MapInstances(IEndpointRouteBuilder app) return instance is null ? Results.NotFound() : Results.Ok(instance.ToDto()); }); + // The instance's state as its own workflow defines it: lifecycle status plus, for a workflow + // that declares an audit record, the record built so far — grouped by the sections the + // definition declared, so the presentation follows the declaration rather than this file + // knowing anything about a particular workflow. + app.MapGet("/workflows/{name}/instances/{id}/state", async ( + string name, + string id, + [FromQuery] string? section, + IInstanceStore instances, + IWorkflowRegistry registry, + IAuditRecordStore auditRecords, + CancellationToken cancellationToken) => + { + WorkflowInstance? instance = await instances.GetAsync(id, cancellationToken).ConfigureAwait(false); + + // A mismatched workflow name is a wrong URL, not a different resource — reading an + // instance through another workflow's route would make the route meaningless. + if (instance is null || !string.Equals(instance.WorkflowName, name, StringComparison.OrdinalIgnoreCase)) + { + return Results.NotFound(); + } + + AuditRecordDefinition? definition = + registry.Resolve(instance.WorkflowName, instance.WorkflowVersion)?.Definition + is IAuditedWorkflowDefinition audited ? audited.AuditRecord : null; + + AuditRecordDocument? document = definition is null + ? null + : await auditRecords.GetAsync(id, cancellationToken).ConfigureAwait(false); + + return Results.Ok(new + { + instance = instance.ToDto(), + audit = BuildAuditState(definition, document, section) + }); + }) + .WithName("GetWorkflowInstanceState") + .WithSummary("Lifecycle status and the workflow's own audit record for one instance."); + app.MapGet("/instances", async ( [FromQuery] string? status, [FromQuery] string? workflow, @@ -464,6 +503,70 @@ private static void MapApprovals(IEndpointRouteBuilder app) }); } + /// + /// Shapes an audit record for the wire. The declared sections drive the shape — every declared + /// section appears, empty ones included, so a caller can see what the workflow has yet to record + /// as readily as what it has. Undeclared entries are ignored the same way the recorder refuses + /// them. Returns null when the workflow keeps no record at all. + /// + private static object? BuildAuditState( + AuditRecordDefinition? definition, AuditRecordDocument? document, string? sectionFilter) + { + if (definition is null) return null; + + string[]? wanted = sectionFilter? + .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + + IEnumerable sections = definition.Sections; + if (wanted is { Length: > 0 }) + { + sections = sections.Where(s => wanted.Contains(s.Kind, StringComparer.OrdinalIgnoreCase)); + } + + var shaped = sections.Select(s => new + { + s.Kind, + s.Description, + s.Multiple, + entries = (document?.Entries ?? []) + .Where(e => string.Equals(e.SectionKind, s.Kind, StringComparison.OrdinalIgnoreCase)) + .OrderBy(e => e.Sequence) + .Select(e => new { e.Key, e.Sequence, e.RecordedUtc, payload = ParseJson(e.PayloadJson) }) + }); + + return new + { + rootKind = definition.RootKind, + definition.Description, + // Null until the workflow opens the record — the shape is known from the definition + // before any run has recorded anything against it. + rootKey = document?.Root.RootKey, + status = document?.Root.Status, + openedUtc = document?.Root.OpenedUtc, + closedUtc = document?.Root.ClosedUtc, + attributes = document is null ? null : ParseJson(document.Root.AttributesJson), + sections = shaped + }; + } + + /// + /// Payloads are stored as JSON text. Re-emitting them as elements keeps the response readable + /// rather than nesting escaped strings inside it. + /// + private static JsonElement? ParseJson(string json) + { + if (string.IsNullOrWhiteSpace(json)) return null; + + try + { + return JsonDocument.Parse(json).RootElement.Clone(); + } + catch (JsonException) + { + return null; + } + } + internal static bool TryParseOutcome(string? value, out ApprovalOutcomeKind outcome) { switch (value?.ToLowerInvariant()) diff --git a/src/Abacus.Run/Api/HostBuilderExtensions.cs b/src/Abacus.Run/Api/HostBuilderExtensions.cs index f7b4b39..4accfd5 100644 --- a/src/Abacus.Run/Api/HostBuilderExtensions.cs +++ b/src/Abacus.Run/Api/HostBuilderExtensions.cs @@ -115,6 +115,7 @@ public static WorkflowHostBuilder AddWorkflowHost( services.TryAddSingleton(); services.TryAddSingleton(); services.TryAddSingleton(); + services.TryAddSingleton(); services.TryAddSingleton(); services.TryAddSingleton(sp => new OverflowCheckpointStore( @@ -194,6 +195,7 @@ internal sealed class WorkflowRunnerFactory : IWorkflowRunnerFactory ApprovalService = _services.GetRequiredService(), GatePolicies = _services.GetRequiredService(), Audit = _services.GetRequiredService(), + AuditRecords = _services.GetRequiredService(), Logs = _services.GetRequiredService(), Services = _services, Options = _services.GetRequiredService>().Value, diff --git a/src/Abacus.Run/Core/WorkflowAuditRecorder.cs b/src/Abacus.Run/Core/WorkflowAuditRecorder.cs new file mode 100644 index 0000000..957eb75 --- /dev/null +++ b/src/Abacus.Run/Core/WorkflowAuditRecorder.cs @@ -0,0 +1,160 @@ +using System.Text.Json; +using Abacus.Run.Abstractions; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Abacus.Run.Core; + +/// +/// Default : validates entries against the workflow's declared +/// record shape, serializes payloads, and writes them to the generic store. One instance per running +/// workflow instance, so the entry sequence is a simple interlocked counter. +/// +/// +/// Storage failures are logged and swallowed. The audit record explains work that has already +/// happened; failing a check because its explanation could not be filed would trade a correct result +/// for a missing one. An undeclared section kind is a programming error in the workflow definition +/// and is logged as a warning rather than throwing, for the same reason. +/// +public sealed class WorkflowAuditRecorder : IWorkflowAuditRecorder +{ + private static readonly JsonSerializerOptions PayloadJson = new(JsonSerializerDefaults.Web) + { + WriteIndented = false + }; + + private readonly IAuditRecordStore _store; + private readonly string _instanceId; + private readonly string _workflowName; + private readonly string _workflowVersion; + private readonly TimeProvider _clock; + private readonly ILogger _logger; + + private int _sequence; + private string _rootKey = string.Empty; + private string _attributesJson = "{}"; + private DateTimeOffset _openedUtc; + + public WorkflowAuditRecorder( + AuditRecordDefinition definition, + IAuditRecordStore store, + string instanceId, + string workflowName, + string workflowVersion, + TimeProvider? clock = null, + ILogger? logger = null) + { + Definition = definition ?? throw new ArgumentNullException(nameof(definition)); + _store = store ?? throw new ArgumentNullException(nameof(store)); + _instanceId = instanceId; + _workflowName = workflowName; + _workflowVersion = workflowVersion; + _clock = clock ?? TimeProvider.System; + _logger = logger ?? NullLogger.Instance; + _openedUtc = _clock.GetUtcNow(); + } + + public AuditRecordDefinition Definition { get; } + + public async ValueTask OpenAsync( + string rootKey, + IReadOnlyDictionary? attributes, + CancellationToken cancellationToken) + { + ArgumentException.ThrowIfNullOrWhiteSpace(rootKey); + + _rootKey = rootKey; + _openedUtc = _clock.GetUtcNow(); + _attributesJson = Serialize(attributes ?? new Dictionary()); + + await SafeAsync( + () => _store.UpsertRootAsync(BuildRoot(AuditRecordStatus.Open, closedUtc: null), cancellationToken), + "open the audit record").ConfigureAwait(false); + } + + public async ValueTask RecordAsync( + string sectionKind, + string? key, + object? payload, + CancellationToken cancellationToken) + { + if (!Definition.Allows(sectionKind)) + { + _logger.LogWarning( + "Audit section '{SectionKind}' is not declared by workflow '{Workflow}'; entry dropped.", + sectionKind, _workflowName); + return; + } + + var entry = new AuditRecordEntry + { + Id = Guid.NewGuid(), + InstanceId = _instanceId, + SectionKind = sectionKind, + Key = key, + PayloadJson = Serialize(payload), + Sequence = Interlocked.Increment(ref _sequence), + RecordedUtc = _clock.GetUtcNow() + }; + + await SafeAsync( + () => _store.AppendAsync(entry, cancellationToken), + $"record audit section '{sectionKind}'").ConfigureAwait(false); + } + + public async ValueTask CloseAsync(string status, CancellationToken cancellationToken) + { + if (_rootKey.Length == 0) + { + // Nothing was opened — closing would write a root with no identity. + _logger.LogDebug("Audit record for instance {InstanceId} was never opened; nothing to close.", _instanceId); + return; + } + + await SafeAsync( + () => _store.UpsertRootAsync( + BuildRoot(status, _clock.GetUtcNow()), cancellationToken), + "close the audit record").ConfigureAwait(false); + } + + private AuditRecordRoot BuildRoot(string status, DateTimeOffset? closedUtc) => new() + { + InstanceId = _instanceId, + WorkflowName = _workflowName, + WorkflowVersion = _workflowVersion, + RootKind = Definition.RootKind, + RootKey = _rootKey, + Status = status, + AttributesJson = _attributesJson, + OpenedUtc = _openedUtc, + ClosedUtc = closedUtc + }; + + private string Serialize(object? payload) + { + if (payload is null) return "null"; + if (payload is string s) return JsonSerializer.Serialize(s, PayloadJson); + + try + { + return JsonSerializer.Serialize(payload, payload.GetType(), PayloadJson); + } + catch (NotSupportedException ex) + { + _logger.LogWarning(ex, "Audit payload of type {Type} could not be serialized.", payload.GetType().Name); + return JsonSerializer.Serialize(new { error = "payload-not-serializable", type = payload.GetType().Name }, PayloadJson); + } + } + + private async ValueTask SafeAsync(Func action, string operation) + { + try + { + await action().ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Failed to {Operation} for instance {InstanceId}.", operation, _instanceId); + } + } +} diff --git a/src/Abacus.Run/Core/WorkflowRunner.cs b/src/Abacus.Run/Core/WorkflowRunner.cs index 5b34a30..84a582b 100644 --- a/src/Abacus.Run/Core/WorkflowRunner.cs +++ b/src/Abacus.Run/Core/WorkflowRunner.cs @@ -25,6 +25,12 @@ public sealed class WorkflowRunnerDependencies public IApprovalService? ApprovalService { get; init; } public IGatePolicyStore? GatePolicies { get; init; } public IAuditStore? Audit { get; init; } + + /// + /// Backing store for workflow audit records. Only consulted for definitions that implement + /// . + /// + public IAuditRecordStore? AuditRecords { get; init; } public ILogStore? Logs { get; init; } public IServiceProvider? Services { get; init; } public WorkflowHostOptions Options { get; init; } = new(); @@ -104,10 +110,15 @@ private async ValueTask ExecuteRunAsync( { var gates = new Dictionary(StringComparer.Ordinal); + // A definition that declares an audit record gets a recorder bound to its declared shape; + // one that doesn't gets null, and the hook costs it nothing. + IWorkflowAuditRecorder? auditRecorder = CreateAuditRecorder(instance, descriptor); + var buildContext = new WorkflowBuildContext( instance.InstanceId, instance.TenantId, instance.WorkflowName, instance.WorkflowVersion, invocation.Attempt, _deps.Services, - (executor, gate) => Attach(executor, gate, instance, invocation, gates)); + (executor, gate) => Attach(executor, gate, instance, invocation, gates, auditRecorder), + auditRecorder); Workflow workflow = await descriptor.Definition.BuildAsync(buildContext, cancellationToken).ConfigureAwait(false); @@ -233,12 +244,39 @@ await PublishAsync(instance, WorkflowEventTypes.WorkflowOutput, return RunOutcome.Completed(output); } + /// + /// Builds the per-instance audit recorder when the definition declares a record shape. Returns + /// null otherwise — auditing is opt-in per workflow, not a tax on every one. + /// + private IWorkflowAuditRecorder? CreateAuditRecorder(WorkflowInstance instance, WorkflowDescriptor descriptor) + { + if (descriptor.Definition is not IAuditedWorkflowDefinition audited) return null; + + if (_deps.AuditRecords is not { } store) + { + _deps.Logger.LogWarning( + "Workflow '{Workflow}' declares an audit record but no IAuditRecordStore is registered; auditing is off.", + instance.WorkflowName); + return null; + } + + return new WorkflowAuditRecorder( + audited.AuditRecord, + store, + instance.InstanceId, + instance.WorkflowName, + instance.WorkflowVersion, + _deps.Clock, + _deps.Logger); + } + private ExecutorBinding Attach( IHostExecutor executor, ApprovalGate gate, WorkflowInstance instance, WorkflowInvocationContext invocation, - Dictionary gates) + Dictionary gates, + IWorkflowAuditRecorder? audit) { gates[executor.Id] = gate; @@ -266,6 +304,7 @@ private ExecutorBinding Attach( _deps.GatePolicies, _deps.Approvals), Approvals = _deps.ApprovalService, Services = _deps.Services, + Audit = audit, ExecutorInvoked = async (executorId, superstep) => { _hostInvocationEvents.Enqueue(executorId); diff --git a/src/Abacus.Run/Persistence/InMemoryAuditRecordStore.cs b/src/Abacus.Run/Persistence/InMemoryAuditRecordStore.cs new file mode 100644 index 0000000..a190a47 --- /dev/null +++ b/src/Abacus.Run/Persistence/InMemoryAuditRecordStore.cs @@ -0,0 +1,66 @@ +using System.Collections.Concurrent; +using Abacus.Run.Abstractions; + +namespace Abacus.Run.Persistence; + +/// +/// Default so a host runs with no storage configured. Records live +/// for the life of the process — deployments that need the audit trail to survive a restart replace +/// this with a durable store. +/// +public sealed class InMemoryAuditRecordStore : IAuditRecordStore +{ + private readonly ConcurrentDictionary _roots = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _entries = + new(StringComparer.Ordinal); + + public ValueTask UpsertRootAsync(AuditRecordRoot root, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(root); + _roots[root.InstanceId] = root; + return ValueTask.CompletedTask; + } + + public ValueTask AppendAsync(AuditRecordEntry entry, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(entry); + + ConcurrentDictionary forInstance = + _entries.GetOrAdd(entry.InstanceId, _ => new ConcurrentDictionary(StringComparer.Ordinal)); + + forInstance[EntryKey(entry.SectionKind, entry.Key)] = entry; + return ValueTask.CompletedTask; + } + + public ValueTask GetAsync(string instanceId, CancellationToken cancellationToken) + { + if (!_roots.TryGetValue(instanceId, out AuditRecordRoot? root)) + { + return ValueTask.FromResult(null); + } + + IReadOnlyList entries = _entries.TryGetValue(instanceId, out var forInstance) + ? [.. forInstance.Values.OrderBy(e => e.Sequence)] + : []; + + return ValueTask.FromResult(new AuditRecordDocument(root, entries)); + } + + public ValueTask> ListAsync( + string workflowName, string? rootKey, string? status, int limit, CancellationToken cancellationToken) + { + IReadOnlyList results = + [ + .. _roots.Values + .Where(r => string.Equals(r.WorkflowName, workflowName, StringComparison.Ordinal)) + .Where(r => rootKey is null || string.Equals(r.RootKey, rootKey, StringComparison.Ordinal)) + .Where(r => status is null || string.Equals(r.Status, status, StringComparison.OrdinalIgnoreCase)) + .OrderByDescending(r => r.OpenedUtc) + .Take(Math.Max(1, limit)) + ]; + + return ValueTask.FromResult(results); + } + + private static string EntryKey(string sectionKind, string? key) => $"{sectionKind}{key ?? string.Empty}"; +} diff --git a/tests/Abacus.Run.IntegrationTests/HostFixture.cs b/tests/Abacus.Run.IntegrationTests/HostFixture.cs index 3fb8215..8cb10cc 100644 --- a/tests/Abacus.Run.IntegrationTests/HostFixture.cs +++ b/tests/Abacus.Run.IntegrationTests/HostFixture.cs @@ -249,6 +249,76 @@ protected override ValueTask ExecuteCoreAsync( } } +/// +/// Declares an audit record, so the generic state endpoint has a workflow to read back. Deliberately +/// unremarkable otherwise — the point is that a definition gets auditing by declaring it, with no +/// framework code that knows what an "order" is. +/// +public sealed class AuditedOrderWorkflow : IWorkflowDefinition, IAuditedWorkflowDefinition +{ + public const string Submission = "submission"; + public const string Step = "step"; + public const string Outcome = "outcome"; + + public string Name => "audited-order"; + public string Version => "1.0.0"; + + public AuditRecordDefinition AuditRecord { get; } = new( + "order", + "One order, as processed.", + [ + new AuditSectionDefinition(Submission, "What was submitted.", Multiple: false), + new AuditSectionDefinition(Step, "One processing step."), + new AuditSectionDefinition(Outcome, "How the run settled.", Multiple: false) + ]); + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + ExecutorBinding validate = context.Node(new Validate("validate")); + ExecutorBinding submit = context.Node(new Submit("submit")); + + return new ValueTask(new WorkflowBuilder(validate) + .AddEdge(validate, submit) + .WithOutputFrom(submit) + .WithName(Name) + .Build()); + } + + private sealed class Validate(string id) : HostExecutor(id) + { + protected override async ValueTask ExecuteCoreAsync( + OrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + if (Runtime.Audit is { } audit) + { + await audit.OpenAsync(input.OrderId, new Dictionary { ["amount"] = input.Amount }, cancellationToken); + await audit.RecordAsync(Submission, null, new { input.OrderId, input.Amount }, cancellationToken); + await audit.RecordAsync(Step, Id, new { accepted = true }, cancellationToken); + } + + return input; + } + } + + private sealed class Submit(string id) : HostExecutor(id) + { + protected override async ValueTask ExecuteCoreAsync( + OrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + var result = new OrderResult(input.OrderId, "submitted"); + + if (Runtime.Audit is { } audit) + { + await audit.RecordAsync(Step, Id, new { result.Status }, cancellationToken); + await audit.RecordAsync(Outcome, null, result, cancellationToken); + await audit.CloseAsync(AuditRecordStatus.Completed, cancellationToken); + } + + return result; + } + } +} + public sealed class HostFixture : WebApplicationFactory { public SideEffectLedger Ledger { get; } = new(); @@ -261,6 +331,7 @@ protected override IHost CreateHost(IHostBuilder builder) services.AddSingleton(sp => new OrderWorkflow(sp.GetRequiredService())); services.AddSingleton(sp => new GatedOrderWorkflow(sp.GetRequiredService())); services.AddSingleton(sp => new TenantOrderWorkflow(sp.GetRequiredService())); + services.AddSingleton(new AuditedOrderWorkflow()); }); return base.CreateHost(builder); diff --git a/tests/Abacus.Run.IntegrationTests/InstanceStateTests.cs b/tests/Abacus.Run.IntegrationTests/InstanceStateTests.cs new file mode 100644 index 0000000..02000bf --- /dev/null +++ b/tests/Abacus.Run.IntegrationTests/InstanceStateTests.cs @@ -0,0 +1,113 @@ +using System.Net; +using System.Net.Http.Json; +using System.Text.Json; +using Abacus.Run.Abstractions; +using FluentAssertions; +using Xunit; + +namespace Abacus.Run.IntegrationTests; + +/// +/// GET /workflows/{name}/instances/{id}/state — lifecycle status plus whatever audit record +/// the workflow declared. The endpoint is generic: it presents the sections the definition declared +/// and knows nothing about any particular workflow. +/// +public class InstanceStateTests : IClassFixture +{ + private readonly HostFixture _fixture; + + public InstanceStateTests(HostFixture fixture) => _fixture = fixture; + + [Fact] + public async Task State_returns_the_audit_record_the_workflow_declared() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, "audited-order", new OrderContext("ORD-STATE-1", 250m)); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement state = await client.GetFromJsonAsync( + $"/workflows/audited-order/instances/{instanceId}/state"); + + state.GetProperty("instance").GetProperty("status").GetString().Should().Be("Completed"); + + JsonElement audit = state.GetProperty("audit"); + audit.GetProperty("rootKind").GetString().Should().Be("order"); + audit.GetProperty("rootKey").GetString().Should().Be("ORD-STATE-1"); + audit.GetProperty("status").GetString().Should().Be("Completed"); + audit.GetProperty("closedUtc").ValueKind.Should().NotBe(JsonValueKind.Null); + audit.GetProperty("attributes").GetProperty("amount").GetDecimal().Should().Be(250m); + + JsonElement[] sections = [.. audit.GetProperty("sections").EnumerateArray()]; + + sections.Select(s => s.GetProperty("kind").GetString()) + .Should().ContainInOrder(["submission", "step", "outcome"], + "the declaration order is the presentation order"); + + JsonElement steps = sections.Single(s => s.GetProperty("kind").GetString() == "step"); + steps.GetProperty("entries").EnumerateArray().Select(e => e.GetProperty("key").GetString()) + .Should().ContainInOrder(["validate", "submit"]); + + JsonElement outcome = sections.Single(s => s.GetProperty("kind").GetString() == "outcome"); + outcome.GetProperty("entries")[0].GetProperty("payload").GetProperty("status").GetString() + .Should().Be("submitted"); + } + + [Fact] + public async Task A_workflow_that_declares_no_record_reports_no_audit() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, "order", new OrderContext("ORD-STATE-2")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement state = await client.GetFromJsonAsync( + $"/workflows/order/instances/{instanceId}/state"); + + state.GetProperty("instance").GetProperty("instanceId").GetString().Should().Be(instanceId); + state.GetProperty("audit").ValueKind.Should().Be(JsonValueKind.Null, + "auditing is opt-in; a workflow that declares nothing records nothing"); + } + + [Fact] + public async Task A_section_filter_narrows_the_record_without_changing_its_shape() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, "audited-order", new OrderContext("ORD-STATE-3")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement state = await client.GetFromJsonAsync( + $"/workflows/audited-order/instances/{instanceId}/state?section=step"); + + JsonElement[] sections = [.. state.GetProperty("audit").GetProperty("sections").EnumerateArray()]; + + sections.Should().ContainSingle(); + sections[0].GetProperty("kind").GetString().Should().Be("step"); + sections[0].GetProperty("entries").GetArrayLength().Should().Be(2); + } + + [Fact] + public async Task Reading_an_instance_through_another_workflows_route_is_not_found() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, "audited-order", new OrderContext("ORD-STATE-4")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + HttpResponseMessage response = await client.GetAsync($"/workflows/order/instances/{instanceId}/state"); + + response.StatusCode.Should().Be(HttpStatusCode.NotFound, + "the workflow segment identifies the resource, so a mismatch is a wrong URL"); + } + + [Fact] + public async Task An_unknown_instance_is_not_found() + { + using HttpClient client = _fixture.CreateClient(); + + HttpResponseMessage response = await client.GetAsync("/workflows/audited-order/instances/does-not-exist/state"); + + response.StatusCode.Should().Be(HttpStatusCode.NotFound); + } +} diff --git a/tests/Abacus.Run.UnitTests/Core/WorkflowAuditRecorderTests.cs b/tests/Abacus.Run.UnitTests/Core/WorkflowAuditRecorderTests.cs new file mode 100644 index 0000000..33c91c8 --- /dev/null +++ b/tests/Abacus.Run.UnitTests/Core/WorkflowAuditRecorderTests.cs @@ -0,0 +1,184 @@ +using System.Text.Json; +using Abacus.Run.Abstractions; +using Abacus.Run.Core; +using Abacus.Run.Persistence; +using FluentAssertions; +using Xunit; + +namespace Abacus.Run.UnitTests.Core; + +/// +/// The audit hook is a framework contract that workflow definitions build on, so its guarantees are +/// tested here rather than only through a workflow that happens to use it. +/// +public class WorkflowAuditRecorderTests +{ + private static readonly AuditRecordDefinition Definition = new( + "order", + "One processed order.", + [ + new AuditSectionDefinition("submission", "What was submitted.", Multiple: false), + new AuditSectionDefinition("step", "One processing step.") + ]); + + private static WorkflowAuditRecorder CreateRecorder(IAuditRecordStore store, string instanceId = "wf-1") + => new(Definition, store, instanceId, "orders", "1.0.0"); + + [Fact] + public async Task Opening_writes_the_root_with_its_key_and_attributes() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + + await recorder.OpenAsync("ORDER-1", new Dictionary { ["channel"] = "web" }, default); + + AuditRecordDocument? document = await store.GetAsync("wf-1", default); + + document.Should().NotBeNull(); + document!.Root.RootKind.Should().Be("order"); + document.Root.RootKey.Should().Be("ORDER-1"); + document.Root.WorkflowName.Should().Be("orders"); + document.Root.Status.Should().Be(AuditRecordStatus.Open); + document.Root.AttributesJson.Should().Contain("web"); + } + + [Fact] + public async Task Entries_are_sequenced_in_the_order_they_are_recorded() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + await recorder.OpenAsync("ORDER-1", null, default); + + await recorder.RecordAsync("submission", null, new { total = 10 }, default); + await recorder.RecordAsync("step", "pick", new { warehouse = "A" }, default); + await recorder.RecordAsync("step", "pack", new { warehouse = "A" }, default); + + AuditRecordDocument document = (await store.GetAsync("wf-1", default))!; + + document.Entries.Select(e => e.Sequence).Should().BeInAscendingOrder(); + document.Entries.Select(e => e.Key).Should().ContainInOrder([null, "pick", "pack"]); + } + + [Fact] + public async Task An_undeclared_section_is_dropped_rather_than_stored() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + await recorder.OpenAsync("ORDER-1", null, default); + + await recorder.RecordAsync("not-declared", null, new { anything = true }, default); + + AuditRecordDocument document = (await store.GetAsync("wf-1", default))!; + document.Entries.Should().BeEmpty("the definition's declared shape is the contract"); + } + + [Fact] + public async Task Re_recording_the_same_section_and_key_replaces_the_entry() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + await recorder.OpenAsync("ORDER-1", null, default); + + await recorder.RecordAsync("step", "pick", new { attempt = 1 }, default); + await recorder.RecordAsync("step", "pick", new { attempt = 2 }, default); + + AuditRecordDocument document = (await store.GetAsync("wf-1", default))!; + + document.Entries.Should().ContainSingle("a retried step corrects its record rather than contradicting it"); + document.Entries[0].PayloadJson.Should().Contain("2"); + } + + [Fact] + public async Task Closing_settles_the_root_and_keeps_the_entries() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + await recorder.OpenAsync("ORDER-1", null, default); + await recorder.RecordAsync("step", "pick", new { ok = true }, default); + + await recorder.CloseAsync(AuditRecordStatus.Completed, default); + + AuditRecordDocument document = (await store.GetAsync("wf-1", default))!; + document.Root.Status.Should().Be(AuditRecordStatus.Completed); + document.Root.ClosedUtc.Should().NotBeNull(); + document.Entries.Should().ContainSingle(); + } + + [Fact] + public async Task A_store_failure_never_propagates_to_the_workflow() + { + // Audit writes explain work that already happened. Failing the work because its explanation + // could not be filed would trade a correct result for a missing one. + WorkflowAuditRecorder recorder = CreateRecorder(new ThrowingAuditRecordStore()); + + Func open = async () => await recorder.OpenAsync("ORDER-1", null, default); + Func record = async () => await recorder.RecordAsync("step", "pick", new { ok = true }, default); + Func close = async () => await recorder.CloseAsync(AuditRecordStatus.Completed, default); + + await open.Should().NotThrowAsync(); + await record.Should().NotThrowAsync(); + await close.Should().NotThrowAsync(); + } + + [Fact] + public async Task Payloads_are_stored_as_json() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + await recorder.OpenAsync("ORDER-1", null, default); + + await recorder.RecordAsync("step", "pick", new { warehouse = "A", items = new[] { 1, 2 } }, default); + + AuditRecordDocument document = (await store.GetAsync("wf-1", default))!; + using JsonDocument payload = JsonDocument.Parse(document.Entries[0].PayloadJson); + + payload.RootElement.GetProperty("warehouse").GetString().Should().Be("A"); + payload.RootElement.GetProperty("items").GetArrayLength().Should().Be(2); + } + + [Fact] + public async Task Closing_without_opening_writes_nothing() + { + var store = new InMemoryAuditRecordStore(); + WorkflowAuditRecorder recorder = CreateRecorder(store); + + await recorder.CloseAsync(AuditRecordStatus.Failed, default); + + (await store.GetAsync("wf-1", default)).Should().BeNull( + "closing a record that was never opened would write a root with no identity"); + } + + [Fact] + public async Task Roots_are_listed_per_workflow_and_filtered_by_status() + { + var store = new InMemoryAuditRecordStore(); + + WorkflowAuditRecorder first = CreateRecorder(store, "wf-1"); + await first.OpenAsync("ORDER-1", null, default); + await first.CloseAsync(AuditRecordStatus.Completed, default); + + WorkflowAuditRecorder second = CreateRecorder(store, "wf-2"); + await second.OpenAsync("ORDER-2", null, default); + + (await store.ListAsync("orders", null, null, 10, default)).Should().HaveCount(2); + (await store.ListAsync("orders", null, AuditRecordStatus.Open, 10, default)).Should().ContainSingle(); + (await store.ListAsync("orders", "ORDER-1", null, 10, default)).Should().ContainSingle(); + (await store.ListAsync("other-workflow", null, null, 10, default)).Should().BeEmpty(); + } + + private sealed class ThrowingAuditRecordStore : IAuditRecordStore + { + public ValueTask UpsertRootAsync(AuditRecordRoot root, CancellationToken cancellationToken) + => throw new InvalidOperationException("storage is down"); + + public ValueTask AppendAsync(AuditRecordEntry entry, CancellationToken cancellationToken) + => throw new InvalidOperationException("storage is down"); + + public ValueTask GetAsync(string instanceId, CancellationToken cancellationToken) + => throw new InvalidOperationException("storage is down"); + + public ValueTask> ListAsync( + string workflowName, string? rootKey, string? status, int limit, CancellationToken cancellationToken) + => throw new InvalidOperationException("storage is down"); + } +} From 03b1726297c3ee05969cfcd07600f0b120481d2b Mon Sep 17 00:00:00 2001 From: Ninja Date: Tue, 11 Aug 2026 00:52:39 +0100 Subject: [PATCH 2/3] feat: add an audit-capable example workflow to the host MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The audit hook had no worked example outside the test fixtures, so the answer to "what does a definition actually have to do" lived only in prose. Workflow example-order is that answer: it declares four sections, opens the record with attributes, files a plan, records one entry per line, and settles both terminal paths. The work it does is deliberately dull — plan, price, total — because the point is the auditing around it, not the domain. Two details are the reason it exists rather than a shorter sample. Line entries are keyed by SKU, so a retried attempt corrects the record instead of appending a contradictory second line. The failure path records the outcome before letting the exception propagate, because a recorder failure is swallowed by design and an explanation filed after the throw is an explanation lost. The host may now define workflows. ArchitectureBoundaryTests carves out the Abacus.Run.Service.Workflows. namespace: extension points outside it still mean the shell has grown behaviour of its own, but a workflow that ships inside the host assembly by convention is a documented consumer of the framework. Test isolation. The host substitutes a SQLite audit store, so every fixture was writing into the deployed database file and inheriting records from previous runs — 600 KB of them had accumulated in the integration bin. Both fixtures now take a temp file each and delete it on dispose, and runtime database files are ignored rather than left to be committed by accident. Verified against a running host, not only the suite: both documented curl invocations return the records shown in the docs. --- .gitignore | 6 + README.md | 28 ++- docs/wiki.md | 66 +++++- src/Abacus.Run.Service/Program.cs | 8 +- .../ExampleOrder/ExampleOrderAuditRecord.cs | 39 ++++ .../ExampleOrder/ExampleOrderWorkflow.cs | 149 +++++++++++++ tests/Abacus.Run.ChaosTests/ChaosFixture.cs | 17 ++ .../ArchitectureBoundaryTests.cs | 20 +- .../ExampleOrderWorkflowTests.cs | 206 ++++++++++++++++++ .../HostFixture.cs | 25 +++ 10 files changed, 543 insertions(+), 21 deletions(-) create mode 100644 src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderAuditRecord.cs create mode 100644 src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderWorkflow.cs create mode 100644 tests/Abacus.Run.IntegrationTests/ExampleOrderWorkflowTests.cs diff --git a/.gitignore b/.gitignore index 92b5358..d75f782 100644 --- a/.gitignore +++ b/.gitignore @@ -422,3 +422,9 @@ docs/ !docs/ docs/* !docs/wiki.md + +# Local SQLite databases written by the host at runtime +data/ +*.db +*.db-shm +*.db-wal diff --git a/README.md b/README.md index 552b42c..fe09642 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,21 @@ The framework default is `InMemoryAuditRecordStore`. `Abacus.Run.Service` displa Core SQLite store via `AddSqliteAuditRecords(configuration)`, configured under `Abacus:AuditRecords:ConnectionString`. +A runnable example ships in the host at +[`src/Abacus.Run.Service/Workflows/ExampleOrder`](src/Abacus.Run.Service/Workflows/ExampleOrder) — +workflow `example-order`. It declares four sections, keys its per-line entries so a retry corrects +the record rather than doubling it, and records the failure before letting it propagate: + +```bash +curl -X POST http://localhost:5000/workflows/example-order/instances \ + -H 'Content-Type: application/json' \ + -d '{"context":{"orderId":"ORD-1","lines":[{"sku":"SKU-A","quantity":2,"unitPrice":10.50}]}}' + +curl http://localhost:5000/workflows/example-order/instances/{id}/state +``` + +Send `"failOnSku": "SKU-A"` in the context to see the failure path and the record it leaves behind. + Full walkthrough: [Workflow audit records](docs/wiki.md#workflow-audit-records). ## Configuration @@ -395,7 +410,7 @@ at startup. | Project | Responsibility | | --- | --- | | `src/Abacus.Run` | Headless framework: workflow runtime, dispatch, executors, middleware, in-memory store defaults, and HTTP API endpoints | -| `src/Abacus.Run.Service` | Deployable host: control-plane UI, SQL Server stores, Redis event bus, and startup wiring | +| `src/Abacus.Run.Service` | Deployable host: control-plane UI, SQL Server stores, Redis event bus, the SQLite audit-record store, startup wiring, and the example workflow | | `tests/Abacus.Run.UnitTests` | Unit coverage for runtime behavior; references the library only | | `tests/Abacus.Run.IntegrationTests` | HTTP, control-plane, and architecture-boundary coverage against the real host | | `tests/Abacus.Run.ChaosTests` | Failure and lifecycle resilience coverage | @@ -409,13 +424,14 @@ src/Abacus.Run/ src/Abacus.Run.Service/ Api/ Infrastructure/ SQL Server stores, Redis bus Core/ Auditing/ audit-record store and migrations Dispatch/ Pages/ control-plane Razor Pages - EventBus/ wwwroot/ control-plane CSS and JS - Executors/ Program.cs - Middlewares/ AbacusServiceCollectionExtensions.cs - Persistence/ + EventBus/ Workflows/ workflow definitions hosted here + Executors/ ExampleOrder/ the audit-capable example workflow + Middlewares/ wwwroot/ control-plane CSS and JS + Persistence/ Program.cs + AbacusServiceCollectionExtensions.cs ``` -The library carries no Razor, MVC, Entity Framework, or Redis dependency; an architecture test in the integration suite enforces this. +The library carries no Razor, MVC, Entity Framework, or Redis dependency; an architecture test in the integration suite enforces this. The same test keeps framework extension points — workflow definitions, host executors, middleware — out of the host shell, carving out only the `Abacus.Run.Service.Workflows.` namespaces where hosted workflows such as the example live. ## Test Coverage diff --git a/docs/wiki.md b/docs/wiki.md index ac88a42..51f94f5 100644 --- a/docs/wiki.md +++ b/docs/wiki.md @@ -142,7 +142,7 @@ The default in-memory stores are suitable for development and tests. They do not | Project | Responsibility | | --- | --- | | `src/Abacus.Run` | Headless framework: workflow contracts, runtime, dispatch, executors, middleware, in-memory store defaults, and the HTTP API endpoints | -| `src/Abacus.Run.Service` | Deployable host: control-plane UI, SQL Server persistence, Redis event bus, and service registration | +| `src/Abacus.Run.Service` | Deployable host: control-plane UI, concrete persistence and event-bus integrations, service registration, and the workflow definitions this deployment runs | | `tests/Abacus.Run.UnitTests` | Focused runtime and store tests; references the library only | | `tests/Abacus.Run.IntegrationTests` | Real host, HTTP endpoint, control-plane, and architecture-boundary tests | | `tests/Abacus.Run.ChaosTests` | Failure and lifecycle resilience tests | @@ -156,10 +156,11 @@ src/Abacus.Run/ src/Abacus.Run.Service/ Api/ Infrastructure/ SQL Server stores, Redis bus Core/ Auditing/ audit-record store and migrations Dispatch/ Pages/ control-plane Razor Pages - EventBus/ wwwroot/ control-plane CSS and JS - Executors/ Program.cs - Middlewares/ AbacusServiceCollectionExtensions.cs - Persistence/ + EventBus/ Workflows/ workflow definitions hosted here + Executors/ / one self-contained folder per workflow + Middlewares/ wwwroot/ control-plane CSS and JS + Persistence/ Program.cs + AbacusServiceCollectionExtensions.cs ``` ### Where the line falls @@ -177,8 +178,10 @@ dependencies. `ArchitectureBoundaryTests` in the integration suite enforces the split: the library must not reference the host, EF Core, Redis, or Razor Pages; every framework contract the host implements must -be a named `SqlServer*` or `Redis*` adapter; and the host must define no workflow definitions, -executors, or middleware of its own. +be a named `SqlServer*` or `Redis*` adapter; and the host must define no framework extension points — +workflow definitions, host executors, middleware — outside a declared +`Abacus.Run.Service.Workflows.` namespace. That carve-out is what lets a workflow ship inside +the host assembly without the rule reading as "the shell may grow behaviour of its own". Workflow authors should normally depend on `Abacus.Run` and its `Abacus.Run.Abstractions` namespace, then register their definitions in the application host. @@ -380,6 +383,53 @@ through the declared sections. Every declared section appears whether or not any recorded into it yet, so a caller reading a run in progress sees what is still outstanding as readily as what is done. +### The worked example + +`src/Abacus.Run.Service/Workflows/ExampleOrder` is a runnable version of everything above, registered +by the host as workflow `example-order`. The work it does is deliberately dull — plan, price each +line, total — because the point is the auditing around it. + +| File | What it shows | +| --- | --- | +| `ExampleOrderAuditRecord.cs` | The declaration in one place, section kinds as constants because they are part of the workflow's contract | +| `ExampleOrderWorkflow.cs` | `OpenAsync` with attributes, a single-entry plan, per-line entries keyed by SKU, and both terminal paths | + +Two details in it are worth copying rather than the shape of the record itself. + +Per-line entries are keyed by SKU. Because re-recording the same `(section, key)` replaces, a retried +attempt corrects the record instead of appending a second, contradictory line. Keying by something +stable about the work — rather than leaving the key null or generating one per attempt — is what buys +that. + +The failure path records the outcome *before* throwing: + +```csharp +var failure = new WorkflowDeadStopException($"Line '{line.Sku}' cannot be priced."); + +if (audit is not null) +{ + await audit.RecordAsync(ExampleOrderAuditRecord.Outcome, null, + new { status = "Failed", failedSku = line.Sku, reason = failure.Message, total }, cancellationToken); + await audit.CloseAsync(AuditRecordStatus.Failed, cancellationToken); +} + +throw failure; +``` + +Start it, then read the record back: + +```bash +curl -X POST http://localhost:5000/workflows/example-order/instances \ + -H 'Content-Type: application/json' \ + -d '{"context":{"orderId":"ORD-1","lines":[{"sku":"SKU-A","quantity":2,"unitPrice":10.50}]}}' + +curl http://localhost:5000/workflows/example-order/instances/{id}/state +``` + +Adding `"failOnSku": "SKU-A"` to the context exercises the failure path. `ExampleOrderWorkflowTests` +in the integration suite covers both, along with the empty-section case and the fact that the host's +SQLite store — not the framework default — is what holds the result. + ## Registering workflows and middleware The API host uses a fluent registration builder: @@ -1015,7 +1065,7 @@ The solution includes several test layers: | Suite | Purpose | | --- | --- | | Unit tests | Contracts, policies, runner behavior, middleware, executors, stores, audit recorder, and control logic | -| Integration tests | Real ASP.NET Core host, HTTP routes, event streams, approvals, instance controls, and the instance state route | +| Integration tests | Real ASP.NET Core host, HTTP routes, event streams, approvals, instance controls, the instance state route, and the example workflow end to end | | Chaos tests | Failure and lifecycle scenarios | | Load tests | Throughput-oriented test project | diff --git a/src/Abacus.Run.Service/Program.cs b/src/Abacus.Run.Service/Program.cs index 8604b8d..5a4d5a1 100644 --- a/src/Abacus.Run.Service/Program.cs +++ b/src/Abacus.Run.Service/Program.cs @@ -3,6 +3,7 @@ using Abacus.Run.Service.Infrastructure.Auditing; using Abacus.Run.Core; using Abacus.Run.Service; +using Abacus.Run.Service.Workflows.ExampleOrder; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -10,7 +11,12 @@ WebApplicationBuilder builder = WebApplication.CreateBuilder(args); builder.Services.AddProblemDetails(); -builder.Services.AddAbacus(builder.Configuration); + +builder.Services + .AddAbacus(builder.Configuration) + // The worked example of the audit hook. It is the only workflow this host ships; a real + // deployment registers its own definitions here the same way. + .AddWorkflow(); // Durable, workflow-agnostic storage for the audit records that workflow definitions declare. // Displaces the framework's in-memory default. diff --git a/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderAuditRecord.cs b/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderAuditRecord.cs new file mode 100644 index 0000000..5beb029 --- /dev/null +++ b/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderAuditRecord.cs @@ -0,0 +1,39 @@ +using Abacus.Run.Abstractions; + +namespace Abacus.Run.Service.Workflows.ExampleOrder; + +/// +/// The shape of this workflow's audit record, declared in one place. +/// +/// +/// Section kinds are written into storage and read back by callers of the state route, so they are +/// part of the workflow's contract rather than incidental strings. Naming them as constants is what +/// keeps a typo at a call site from silently producing an undeclared section the recorder drops. +/// +public static class ExampleOrderAuditRecord +{ + public const string RootKind = "order"; + + /// What the caller asked for. One per run. + public const string Submission = "submission"; + + /// The plan formed before any line was priced. One per run. + public const string Plan = "plan"; + + /// One priced line, keyed by SKU. Many per run. + public const string Line = "line"; + + /// How the run settled — including a failure, when it failed. One per run. + public const string Outcome = "outcome"; + + public static readonly AuditRecordDefinition Definition = new( + RootKind, + "One order, as this workflow processed it: what was asked for, how it was planned, what each " + + "line cost, and how the run settled.", + [ + new AuditSectionDefinition(Submission, "The order as submitted.", Multiple: false), + new AuditSectionDefinition(Plan, "The pricing plan formed before acting.", Multiple: false), + new AuditSectionDefinition(Line, "One priced order line, keyed by SKU."), + new AuditSectionDefinition(Outcome, "The settled total, or the failure that stopped the run.", Multiple: false) + ]); +} diff --git a/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderWorkflow.cs b/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderWorkflow.cs new file mode 100644 index 0000000..b9afb29 --- /dev/null +++ b/src/Abacus.Run.Service/Workflows/ExampleOrder/ExampleOrderWorkflow.cs @@ -0,0 +1,149 @@ +using Abacus.Run.Abstractions; +using Microsoft.Agents.AI.Workflows; + +namespace Abacus.Run.Service.Workflows.ExampleOrder; + +/// What one run is asked to price. FailOnSku exists so the failure path is reachable. +public sealed record ExampleOrderContext( + string OrderId = "ORD-1", + IReadOnlyList? Lines = null, + string? FailOnSku = null) +{ + public IReadOnlyList Lines { get; init; } = Lines ?? []; +} + +public sealed record ExampleOrderLine(string Sku, int Quantity, decimal UnitPrice); + +public sealed record ExampleOrderResult(string OrderId, decimal Total, int LineCount); + +/// +/// A worked example of the framework's audit hook, and the only workflow this host ships. +/// +/// +/// It is deliberately dull work — plan, price each line, total — because the point is the auditing +/// around it, not the domain. Read it as the answer to "what does a definition have to do to get an +/// audit record": implement , declare a shape, and call the +/// recorder the runtime hands each node. Nothing in the framework knows what an order is. +/// +public sealed class ExampleOrderWorkflow + : IWorkflowDefinition, IAuditedWorkflowDefinition +{ + public string Name => "example-order"; + public string Version => "1.0.0"; + + public AuditRecordDefinition AuditRecord => ExampleOrderAuditRecord.Definition; + + public ValueTask BuildAsync(WorkflowBuildContext context, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(context); + + ExecutorBinding plan = context.Node(new PlanOrder("plan")); + ExecutorBinding price = context.Node(new PriceLines("price")); + + return new ValueTask(new WorkflowBuilder(plan) + .AddEdge(plan, price) + .WithOutputFrom(price) + .WithName(Name) + .Build()); + } + + /// Opens the record and files what it was asked to do, before doing any of it. + private sealed class PlanOrder(string id) : HostExecutor(id) + { + protected override async ValueTask ExecuteCoreAsync( + ExampleOrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + // Null whenever the workflow declares no record, so every call site is guarded. Here it is + // never null in practice — the guard is the habit an example should teach. + if (Runtime.Audit is { } audit) + { + // Opening is idempotent per instance, so a retried attempt reuses the same root + // rather than starting a second record for the same run. + await audit.OpenAsync( + input.OrderId, + new Dictionary + { + ["lineCount"] = input.Lines.Count, + ["attempt"] = Runtime.Attempt + }, + cancellationToken).ConfigureAwait(false); + + await audit.RecordAsync( + ExampleOrderAuditRecord.Submission, null, input, cancellationToken).ConfigureAwait(false); + + await audit.RecordAsync( + ExampleOrderAuditRecord.Plan, + null, + new { steps = new[] { "price-lines", "total" }, skus = input.Lines.Select(l => l.Sku) }, + cancellationToken).ConfigureAwait(false); + } + + return input; + } + } + + /// + /// Prices each line, filing one entry per SKU. Keying entries by SKU is what makes a retry + /// correct the record instead of doubling it — a re-record of the same (section, key) replaces. + /// + private sealed class PriceLines(string id) : HostExecutor(id) + { + protected override async ValueTask ExecuteCoreAsync( + ExampleOrderContext input, IWorkflowContext context, CancellationToken cancellationToken) + { + IWorkflowAuditRecorder? audit = Runtime.Audit; + decimal total = 0m; + + foreach (ExampleOrderLine line in input.Lines) + { + if (string.Equals(line.Sku, input.FailOnSku, StringComparison.OrdinalIgnoreCase)) + { + var failure = new WorkflowDeadStopException($"Line '{line.Sku}' cannot be priced."); + + // Recorded *before* the throw. A recorder failure is swallowed by design, so the + // record is only useful if it is written while the run can still write it — + // filing the explanation after letting the exception go loses the explanation. + if (audit is not null) + { + await audit.RecordAsync( + ExampleOrderAuditRecord.Outcome, + null, + new { status = "Failed", failedSku = line.Sku, reason = failure.Message, total }, + cancellationToken).ConfigureAwait(false); + + await audit.CloseAsync(AuditRecordStatus.Failed, cancellationToken).ConfigureAwait(false); + } + + throw failure; + } + + decimal lineTotal = line.Quantity * line.UnitPrice; + total += lineTotal; + + if (audit is not null) + { + await audit.RecordAsync( + ExampleOrderAuditRecord.Line, + line.Sku, + new { line.Sku, line.Quantity, line.UnitPrice, lineTotal }, + cancellationToken).ConfigureAwait(false); + } + } + + var result = new ExampleOrderResult(input.OrderId, total, input.Lines.Count); + + if (audit is not null) + { + await audit.RecordAsync( + ExampleOrderAuditRecord.Outcome, + null, + new { status = "Completed", result.Total, result.LineCount }, + cancellationToken).ConfigureAwait(false); + + await audit.CloseAsync(AuditRecordStatus.Completed, cancellationToken).ConfigureAwait(false); + } + + return result; + } + } +} diff --git a/tests/Abacus.Run.ChaosTests/ChaosFixture.cs b/tests/Abacus.Run.ChaosTests/ChaosFixture.cs index ef8673a..5a2261c 100644 --- a/tests/Abacus.Run.ChaosTests/ChaosFixture.cs +++ b/tests/Abacus.Run.ChaosTests/ChaosFixture.cs @@ -76,11 +76,19 @@ public async Task RestartAsync(ReplicaHandle replica) await fresh.StartAsync(); } + /// + /// One audit database for the whole fixture, shared by its replicas the way the other stores are + /// and separate from the deployed file, which every fixture would otherwise write into. + /// + private readonly string _auditDatabasePath = + Path.Combine(Path.GetTempPath(), $"abacus-chaos-audit-{Guid.NewGuid():N}.db"); + private ReplicaHandle CreateReplica(string replicaId) { var factory = new WebApplicationFactory() .WithWebHostBuilder(builder => { + builder.UseSetting("Abacus:AuditRecords:ConnectionString", $"Data Source={_auditDatabasePath}"); builder.ConfigureServices(services => { // Share stores across replicas. @@ -105,6 +113,15 @@ public async ValueTask DisposeAsync() { await replica.DisposeAsync(); } + + // Best-effort: a file left behind is untidy, not a test failure. + try + { + if (File.Exists(_auditDatabasePath)) File.Delete(_auditDatabasePath); + } + catch (IOException) + { + } } } diff --git a/tests/Abacus.Run.IntegrationTests/ArchitectureBoundaryTests.cs b/tests/Abacus.Run.IntegrationTests/ArchitectureBoundaryTests.cs index 3730e7b..6a6e8f4 100644 --- a/tests/Abacus.Run.IntegrationTests/ArchitectureBoundaryTests.cs +++ b/tests/Abacus.Run.IntegrationTests/ArchitectureBoundaryTests.cs @@ -105,10 +105,13 @@ public void Every_framework_contract_the_host_implements_is_an_infrastructure_ad } [Fact] - public void The_host_defines_no_framework_extension_points() + public void The_host_defines_no_framework_extension_points_outside_declared_workflows() { - // Workflow definitions, executors and middleware are authored by consumers against the - // library. Finding one here would mean the shell had grown behaviour of its own. + // Framework extension points (workflow definitions, host executors, middleware) belong to + // consumers of Abacus.Run — not to the shell that wires infrastructure. The rule carves out + // workflows that ship inside the Service assembly by convention: anything under the + // "…Workflows." namespace is a documented consumer of the framework, not shell code + // that has drifted. Type[] extensionPoints = [ typeof(Abacus.Run.Abstractions.IWorkflowDefinition), @@ -117,11 +120,16 @@ public void The_host_defines_no_framework_extension_points() typeof(Abacus.Run.Abstractions.Middleware.IExecutorMiddleware) ]; - Host.GetTypes() + string[] offenders = Host.GetTypes() .Where(t => t is { IsClass: true, IsAbstract: false }) .Where(t => extensionPoints.Any(e => e.IsAssignableFrom(t))) - .Select(t => t.Name) - .Should().BeEmpty(); + .Where(t => t.Namespace is null + || !t.Namespace.StartsWith("Abacus.Run.Service.Workflows.", StringComparison.Ordinal)) + .Select(t => t.FullName!) + .ToArray(); + + offenders.Should().BeEmpty( + "extension points outside a declared workflow namespace would mean the shell had grown behaviour of its own"); } [Fact] diff --git a/tests/Abacus.Run.IntegrationTests/ExampleOrderWorkflowTests.cs b/tests/Abacus.Run.IntegrationTests/ExampleOrderWorkflowTests.cs new file mode 100644 index 0000000..51a336b --- /dev/null +++ b/tests/Abacus.Run.IntegrationTests/ExampleOrderWorkflowTests.cs @@ -0,0 +1,206 @@ +using System.Net.Http.Json; +using System.Text.Json; +using Abacus.Run.Abstractions; +using Abacus.Run.Service.Infrastructure.Auditing; +using Abacus.Run.Service.Workflows.ExampleOrder; +using FluentAssertions; +using Xunit; + +namespace Abacus.Run.IntegrationTests; + +/// +/// The example workflow the host ships, driven end to end through the real API. +/// +/// +/// covers the state route against a fixture-local workflow, which +/// proves the endpoint is generic. These tests cover the other half: that a workflow registered by +/// the host in the ordinary way — declaring a record, calling the recorder from its executors — +/// produces the record a reader expects, on the success path and the failure path, and that the +/// host's durable store is what backs it. +/// +public class ExampleOrderWorkflowTests : IClassFixture +{ + private const string Workflow = "example-order"; + + private readonly HostFixture _fixture; + + public ExampleOrderWorkflowTests(HostFixture fixture) => _fixture = fixture; + + private static ExampleOrderContext Order(string orderId, string? failOnSku = null) => new( + orderId, + [ + new ExampleOrderLine("SKU-A", 2, 10.50m), + new ExampleOrderLine("SKU-B", 1, 99.00m) + ], + failOnSku); + + [Fact] + public async Task The_example_workflow_is_registered_and_discoverable() + { + using HttpClient client = _fixture.CreateClient(); + + JsonElement catalog = await client.GetFromJsonAsync("/workflows"); + + catalog.EnumerateArray().Select(w => w.GetProperty("name").GetString()) + .Should().Contain(Workflow, "the host registers the example in Program.cs"); + } + + [Fact] + public async Task A_completed_run_records_every_declared_section() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, Workflow, Order("ORD-EX-1")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement audit = await ReadAuditAsync(client, instanceId); + + audit.GetProperty("rootKind").GetString().Should().Be("order"); + audit.GetProperty("rootKey").GetString().Should().Be("ORD-EX-1", + "the root key is the workflow's own identifier for the thing being audited"); + audit.GetProperty("status").GetString().Should().Be(AuditRecordStatus.Completed); + audit.GetProperty("closedUtc").ValueKind.Should().NotBe(JsonValueKind.Null); + audit.GetProperty("attributes").GetProperty("lineCount").GetInt32().Should().Be(2); + + JsonElement[] sections = [.. audit.GetProperty("sections").EnumerateArray()]; + + sections.Select(s => s.GetProperty("kind").GetString()) + .Should().ContainInOrder(["submission", "plan", "line", "outcome"], + "the declaration order is the presentation order"); + + Section(sections, ExampleOrderAuditRecord.Submission) + .GetProperty("entries")[0].GetProperty("payload").GetProperty("orderId").GetString() + .Should().Be("ORD-EX-1"); + + Section(sections, ExampleOrderAuditRecord.Plan) + .GetProperty("entries")[0].GetProperty("payload").GetProperty("skus") + .EnumerateArray().Select(s => s.GetString()) + .Should().Equal("SKU-A", "SKU-B"); + + JsonElement outcome = Section(sections, ExampleOrderAuditRecord.Outcome); + outcome.GetProperty("entries")[0].GetProperty("payload").GetProperty("total").GetDecimal() + .Should().Be(120.00m, "2 x 10.50 plus 1 x 99.00"); + } + + [Fact] + public async Task Line_entries_are_keyed_by_sku_and_ordered_by_sequence() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, Workflow, Order("ORD-EX-2")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement audit = await ReadAuditAsync(client, instanceId); + JsonElement lines = Section([.. audit.GetProperty("sections").EnumerateArray()], ExampleOrderAuditRecord.Line); + + lines.GetProperty("multiple").GetBoolean().Should().BeTrue(); + + JsonElement[] entries = [.. lines.GetProperty("entries").EnumerateArray()]; + + entries.Select(e => e.GetProperty("key").GetString()) + .Should().ContainInOrder(["SKU-A", "SKU-B"], "entries come back in recorded order"); + + entries.Select(e => e.GetProperty("sequence").GetInt32()) + .Should().BeInAscendingOrder().And.OnlyHaveUniqueItems(); + + entries[0].GetProperty("payload").GetProperty("lineTotal").GetDecimal().Should().Be(21.00m); + } + + [Fact] + public async Task A_failed_run_records_the_failure_that_stopped_it() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, Workflow, Order("ORD-EX-3", failOnSku: "SKU-B")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.DeadStopped); + + JsonElement audit = await ReadAuditAsync(client, instanceId); + + audit.GetProperty("status").GetString().Should().Be(AuditRecordStatus.Failed); + + JsonElement[] sections = [.. audit.GetProperty("sections").EnumerateArray()]; + + JsonElement outcome = Section(sections, ExampleOrderAuditRecord.Outcome).GetProperty("entries")[0] + .GetProperty("payload"); + outcome.GetProperty("status").GetString().Should().Be("Failed"); + outcome.GetProperty("failedSku").GetString().Should().Be("SKU-B", + "the record is written before the exception propagates, so it explains the failure it caused"); + + Section(sections, ExampleOrderAuditRecord.Line).GetProperty("entries") + .EnumerateArray().Select(e => e.GetProperty("key").GetString()) + .Should().ContainSingle("the line priced before the failure is still on the record") + .Which.Should().Be("SKU-A"); + } + + [Fact] + public async Task Sections_untouched_by_a_run_come_back_empty_rather_than_missing() + { + using HttpClient client = _fixture.CreateClient(); + + // No lines: the workflow opens the record, plans, and settles without pricing anything. + string instanceId = await _fixture.StartAsync( + client, Workflow, new ExampleOrderContext("ORD-EX-4", Lines: [])); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement audit = await ReadAuditAsync(client, instanceId); + JsonElement[] sections = [.. audit.GetProperty("sections").EnumerateArray()]; + + sections.Should().HaveCount(4, "every declared section is presented, recorded into or not"); + Section(sections, ExampleOrderAuditRecord.Line).GetProperty("entries").GetArrayLength() + .Should().Be(0, "an outstanding section is visible as empty, not absent"); + } + + [Fact] + public async Task A_section_filter_narrows_the_example_record() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, Workflow, Order("ORD-EX-5")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + JsonElement state = await client.GetFromJsonAsync( + $"/workflows/{Workflow}/instances/{instanceId}/state?section=plan,outcome"); + + state.GetProperty("audit").GetProperty("sections").EnumerateArray() + .Select(s => s.GetProperty("kind").GetString()) + .Should().Equal("plan", "outcome"); + } + + [Fact] + public async Task The_record_is_held_by_the_hosts_durable_store() + { + using HttpClient client = _fixture.CreateClient(); + + string instanceId = await _fixture.StartAsync(client, Workflow, Order("ORD-EX-6")); + await _fixture.WaitForStatusAsync(instanceId, InstanceStatus.Completed); + + var store = _fixture.Resolve(); + store.Should().BeOfType( + "the host displaces the framework's in-memory default"); + + // Read straight from the store rather than through the API: the record outlives the request + // that produced it, which is the whole point of substituting a durable store. + AuditRecordDocument? document = await store.GetAsync(instanceId, default); + + document.Should().NotBeNull(); + document!.Root.RootKey.Should().Be("ORD-EX-6"); + document.Root.WorkflowName.Should().Be(Workflow); + document.Entries.Should().HaveCount(5, "submission, plan, two lines, outcome"); + + IReadOnlyList roots = await store.ListAsync(Workflow, "ORD-EX-6", null, 10, default); + roots.Should().ContainSingle().Which.InstanceId.Should().Be(instanceId); + } + + private static async Task ReadAuditAsync(HttpClient client, string instanceId) + { + JsonElement state = await client.GetFromJsonAsync( + $"/workflows/{Workflow}/instances/{instanceId}/state"); + + JsonElement audit = state.GetProperty("audit"); + audit.ValueKind.Should().NotBe(JsonValueKind.Null, "the example workflow declares a record"); + return audit; + } + + private static JsonElement Section(JsonElement[] sections, string kind) => + sections.Single(s => s.GetProperty("kind").GetString() == kind); +} diff --git a/tests/Abacus.Run.IntegrationTests/HostFixture.cs b/tests/Abacus.Run.IntegrationTests/HostFixture.cs index 8cb10cc..f6bfc24 100644 --- a/tests/Abacus.Run.IntegrationTests/HostFixture.cs +++ b/tests/Abacus.Run.IntegrationTests/HostFixture.cs @@ -337,9 +337,18 @@ protected override IHost CreateHost(IHostBuilder builder) return base.CreateHost(builder); } + /// + /// The host substitutes a SQLite audit-record store for the framework default, so without an + /// override every fixture would share the deployed database file and inherit records from + /// previous runs. One file per fixture, deleted on dispose, keeps that state out of the suite. + /// + private readonly string _auditDatabasePath = + Path.Combine(Path.GetTempPath(), $"abacus-audit-{Guid.NewGuid():N}.db"); + protected override void ConfigureWebHost(IWebHostBuilder builder) { builder.UseSetting("WorkflowHost:Approvals:SweepIntervalSeconds", "1"); + builder.UseSetting("Abacus:AuditRecords:ConnectionString", $"Data Source={_auditDatabasePath}"); builder.ConfigureServices(services => { services.AddHttpClient(client => @@ -397,4 +406,20 @@ public async Task StartAsync( JsonElement body = await response.Content.ReadFromJsonAsync(); return body.GetProperty("instanceId").GetString()!; } + + protected override void Dispose(bool disposing) + { + base.Dispose(disposing); + + if (!disposing) return; + + // Best-effort: a file left behind is untidy, not a test failure. + try + { + if (File.Exists(_auditDatabasePath)) File.Delete(_auditDatabasePath); + } + catch (IOException) + { + } + } } From 1e61867379783eadcdeb872e1917653f370b30c9 Mon Sep 17 00:00:00 2001 From: Ninja Date: Tue, 11 Aug 2026 01:06:37 +0100 Subject: [PATCH 3/3] docs: record the attribution rule for this repository Commits here are authored by the repository owner. The tooling used to produce a change is not part of the change's history, so no co-author trailer, no generated-with footer on pull requests, and no tool attribution in code or docs. --- CLAUDE.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..6c6e29f --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,13 @@ +# Abacus Run — working notes + +## Attribution + +Do not add Claude, Anthropic, or Claude Code as an author, co-author, or contributor to anything +in this repository. Concretely: + +- No `Co-Authored-By: Claude ...` trailer on commits. +- No "Generated with Claude Code" footer on pull request bodies or issue comments. +- No tool attribution in code comments, docs, or changelogs. + +Commits are authored by the repository owner. The tooling used to produce a change is not part of +the change's history.