From 799564e51c79c408bfab516fdd4958cfb7fdb3af Mon Sep 17 00:00:00 2001 From: fabbrik <22822543+fabbrik@users.noreply.github.com> Date: Fri, 18 Sep 2026 15:33:19 -0300 Subject: [PATCH] feat: commit audited lifecycle changes atomically Add a lifecycle commit and history read to IExperienceRecordStore, and a Core ExperienceLifecycleService that decides the minimal transitions (validate, quarantine, revoke) and stamps the event. The PostgreSQL adapter appends the event and updates the record projection in one READ COMMITTED transaction, idempotent by event ID, guarded by expected revision and by the record's stored status, with an append-only lifecycle_events table added by migration 0002. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 15 +- .../ExperienceRecordStore.cs | 129 +++- .../LifecycleEvent.cs | 11 +- .../Lifecycle/ExperienceLifecycleService.cs | 141 ++++ .../Lifecycle/LifecycleResults.cs | 110 +++ .../AgentExperience.Storage.Postgres.csproj | 1 + .../ExperienceRecordValidator.cs | 52 ++ .../ExperienceSchemaMigrator.cs | 5 +- .../0002_create_lifecycle_events.sql | 56 ++ .../PostgresExperienceRecordSchema.cs | 5 +- .../PostgresExperienceRecordStore.cs | 408 ++++++++++- .../README.md | 92 ++- .../ContractShapeTests.cs | 81 ++- .../ExperienceLifecycleServiceTests.cs | 319 +++++++++ .../OfflineStoreTests.cs | 138 +++- .../PostgresLifecycleCommitTests.cs | 647 ++++++++++++++++++ .../TestRecords.cs | 18 + 17 files changed, 2192 insertions(+), 36 deletions(-) create mode 100644 src/AgentExperience.Core/Lifecycle/ExperienceLifecycleService.cs create mode 100644 src/AgentExperience.Core/Lifecycle/LifecycleResults.cs create mode 100644 src/AgentExperience.Storage.Postgres/Migrations/0002_create_lifecycle_events.sql create mode 100644 tests/AgentExperience.Core.Tests/ExperienceLifecycleServiceTests.cs create mode 100644 tests/AgentExperience.Storage.Postgres.Tests/PostgresLifecycleCommitTests.cs diff --git a/README.md b/README.md index 2b85a7f..0ef8ec8 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ AgentExperience.NET captures what an AI agent actually tried, verifies whether it worked, and turns the result into an auditable lesson that future runs can reuse safely. It sits between [Microsoft Agent Framework](https://github.com/microsoft/agent-framework) (MAF) execution and durable storage, without replacing either. -> **Status: early development.** Epic 1 (capture and explain agent experience) is implemented and tested. Epic 2 has started: Experience Records can be stored in PostgreSQL. Retrieval, injection, and governance are planned (see [Roadmap](#roadmap)). Nothing is published to NuGet yet, and APIs may change. +> **Status: early development.** Epic 1 (capture and explain agent experience) is implemented and tested. Epic 2 has started: Experience Records can be stored in PostgreSQL and moved through their lifecycle with atomic, audited commits. Retrieval, injection, and governance are planned (see [Roadmap](#roadmap)). Nothing is published to NuGet yet, and APIs may change. ## Why @@ -31,6 +31,7 @@ AgentExperience.NET records observable evidence (tool calls, results, errors, ve | Auditable, template-based reflections traceable to evidence IDs | `AgentExperience.Core` | | MAF adapter: captures ordinary, streaming, failed, and cancelled runs plus tool calls, without altering results | `AgentExperience.MicrosoftAgentFramework` | | PostgreSQL Experience Record store: create, get, and scoped query; host authorization checked before database access; exact scope matching in SQL | `AgentExperience.Storage.Postgres` | +| Atomic audited lifecycle commits: the event and the record's projection in one transaction, idempotent by event ID, revision-checked, with append-only history | `AgentExperience.Core`, `AgentExperience.Storage.Postgres` | | Journaled schema migrations: embedded scripts applied once, one transaction per script, serialized across processes by an advisory lock | `AgentExperience.Storage.Postgres` | ## Quick look @@ -66,12 +67,12 @@ See the [adapter README](src/AgentExperience.MicrosoftAgentFramework/README.md) ``` src/ AgentExperience.Abstractions/ domain contracts and ports (BCL only) - AgentExperience.Core/ sanitization, capture, verification, reflection + AgentExperience.Core/ sanitization, capture, verification, reflection, lifecycle transitions AgentExperience.MicrosoftAgentFramework/ MAF adapter (pinned Microsoft.Agents.AI 1.20.0) AgentExperience.Storage.Postgres/ PostgreSQL Experience Record store and schema migrator (pinned Npgsql 10.0.3, dbup-postgresql 7.0.1, dbup-core 6.1.1) tests/ AgentExperience.Abstractions.Tests/ contract and dependency-boundary tests - AgentExperience.Core.Tests/ sanitizer, capture, verification, reflection tests + AgentExperience.Core.Tests/ sanitizer, capture, verification, reflection, lifecycle tests AgentExperience.MicrosoftAgentFramework.Tests/ real ChatClientAgent runs against a scripted fake model AgentExperience.Storage.Postgres.Tests/ store tests, mostly against a PostgreSQL container AgentExperience.CompatibilityProof/ executable proofs for MAF hooks, context providers, pgvector, redaction @@ -89,17 +90,17 @@ dotnet build dotnet test ``` -Unit and MAF adapter tests run in memory, with no network, database, or model credentials. `AgentExperience.CompatibilityProof` and the `PostgresExperienceRecordStoreTests` and `ExperienceSchemaMigratorTests` in `AgentExperience.Storage.Postgres.Tests` start a PostgreSQL/pgvector container through Testcontainers, so they need Docker. If Testcontainers' Ryuk container fails to start under your local Docker setup, set `TESTCONTAINERS_RYUK_DISABLED=true`. To skip the container-backed tests: +Unit and MAF adapter tests run in memory, with no network, database, or model credentials. `AgentExperience.CompatibilityProof` and the `PostgresExperienceRecordStoreTests`, `PostgresLifecycleCommitTests`, and `ExperienceSchemaMigratorTests` in `AgentExperience.Storage.Postgres.Tests` start a PostgreSQL/pgvector container through Testcontainers, so they need Docker. If Testcontainers' Ryuk container fails to start under your local Docker setup, set `TESTCONTAINERS_RYUK_DISABLED=true`. To skip the container-backed tests: ```bash -dotnet test --filter "FullyQualifiedName!~CompatibilityProof&FullyQualifiedName!~PostgresExperienceRecordStoreTests&FullyQualifiedName!~ExperienceSchemaMigratorTests" +dotnet test --filter "FullyQualifiedName!~CompatibilityProof&FullyQualifiedName!~PostgresExperienceRecordStoreTests&FullyQualifiedName!~PostgresLifecycleCommitTests&FullyQualifiedName!~ExperienceSchemaMigratorTests" ``` ## Roadmap 1. **Capture and explain agent experience** ✅ contracts, sanitization, capture, verification, reflection, MAF adapter -2. **Reuse relevant experience:** PostgreSQL persistence (Experience Record store in place), hybrid text and vector retrieval, historical-reference injection into MAF -3. **Govern experience safely:** sharing grants, audited lifecycle transitions, evidence-based confidence updates +2. **Reuse relevant experience:** PostgreSQL persistence and atomic audited lifecycle commits (in place), hybrid text and vector retrieval, historical-reference injection into MAF +3. **Govern experience safely:** sharing grants, the remaining lifecycle transitions, evidence-based confidence updates 4. **Operate and measure the learning loop:** OpenTelemetry instrumentation, an end-to-end demo, measured reuse against a baseline, data deletion and expiry Full requirements and acceptance criteria are in [`_sdlc/planning-artifacts/epics.md`](_sdlc/planning-artifacts/epics.md). diff --git a/src/AgentExperience.Abstractions/ExperienceRecordStore.cs b/src/AgentExperience.Abstractions/ExperienceRecordStore.cs index 91f48b7..fd630f5 100644 --- a/src/AgentExperience.Abstractions/ExperienceRecordStore.cs +++ b/src/AgentExperience.Abstractions/ExperienceRecordStore.cs @@ -57,6 +57,73 @@ Task QueryAsync( AuthorizationContext authorization, ExperienceRecordQuery query, CancellationToken cancellationToken); + + /// + /// Appends and updates the record's projection + /// (, , + /// ) in one transaction: both writes commit together or + /// neither does. The store persists the decision exactly as given and never derives a status, a + /// score, or a counter of its own -- deciding which transition is legal belongs to Core. + /// + /// + /// + /// Idempotency. is the idempotency key. Replaying an + /// event whose stored fields (including its scope) are identical returns the original outcome -- + /// with the revision that commit produced -- and + /// writes nothing. A stored with any differing field is + /// , whichever scope owns it, and writes nothing. + /// + /// + /// Concurrency. must equal the record's + /// current . A successful commit sets the revision to + /// + 1. Any other value is + /// and writes nothing, so two commits racing + /// from the same revision never both apply. + /// + /// + /// Prior-status guard. When is non-null it must + /// also equal the record's stored , matched in the same + /// statement as the revision. That is what keeps Core's transition table enforced against real + /// state rather than against what the caller asserted, and keeps a stored event from recording a + /// prior status the record never had. A mismatch is + /// , carries the stored status, and writes + /// nothing. A (a record's first + /// event) skips the status match. + /// + /// + /// What the host has established the caller may do. + /// The exact request scope the record must lie in. Never treated as authority. + /// The transition Core decided, already stamped. + /// Cancels the operation. + /// + /// , , + /// , + /// (missing, or in another scope), + /// , , or + /// . + /// + Task CommitLifecycleEventAsync( + AuthorizationContext authorization, + Scope scope, + LifecycleEvent lifecycleEvent, + CancellationToken cancellationToken); + + /// + /// Reads one record's lifecycle history within exactly : its current + /// plus every appended event, oldest first. A record that + /// exists in a different scope is indistinguishable from a missing one + /// (). Events are never deleted or rewritten. + /// + /// What the host has established the caller may do. + /// The exact request scope to read within. + /// The record whose history to read. Must not be . + /// Cancels the operation. + /// (possibly with no events), , , or . + Task GetHistoryAsync( + AuthorizationContext authorization, + Scope scope, + Guid experienceId, + CancellationToken cancellationToken); } /// @@ -100,8 +167,33 @@ public enum ExperienceStoreOutcome /// The request was malformed. See the result's validation errors. No storage was accessed. Invalid, - /// A record with the same ID already exists in some scope. The stored record is unchanged and not revealed. + /// + /// A record with the same ID already exists in some scope, or a lifecycle event with the same + /// is already stored with differing fields. Nothing was + /// written and the stored state is unchanged and not revealed. + /// Conflict, + + /// + /// A lifecycle event and its projection update were committed together. The record's + /// is now + 1. + /// An identical replay reports this same outcome without writing again. + /// + Committed, + + /// + /// The lifecycle event's did not equal the record's + /// current , so newer state was not overwritten. Nothing + /// was written. + /// + StaleRevision, + + /// + /// The lifecycle event's did not equal the record's stored + /// , so the transition was decided against state the record was + /// not in. Nothing was written, and the result carries the stored status to re-decide against. + /// + StatusMismatch, } /// @@ -142,6 +234,41 @@ public sealed record ExperienceRecordQueryResult( IReadOnlyList Records, IReadOnlyList Errors); +/// +/// The result of . +/// +/// What happened. +/// +/// The record's after a +/// commit (or after the original commit, when this call +/// was an identical replay); the record's current revision on +/// , so the caller can retry against it; otherwise 0. +/// +/// +/// The record's stored when is +/// , so the caller can re-decide the transition +/// against the state the record is actually in; otherwise . +/// +/// Every validation error when is ; otherwise empty. +public sealed record ExperienceLifecycleCommitResult( + ExperienceStoreOutcome Outcome, + long Revision, + ExperienceStatus? CurrentStatus, + IReadOnlyList Errors); + +/// +/// The result of . +/// +/// What happened. +/// The record's current when is ; otherwise 0. +/// The record's lifecycle events, oldest first, when is ; otherwise empty. +/// Every validation error when is ; otherwise empty. +public sealed record ExperienceRecordHistoryResult( + ExperienceStoreOutcome Outcome, + long Revision, + IReadOnlyList Events, + IReadOnlyList Errors); + /// /// Thrown by an implementation when storage infrastructure /// fails (database unavailable, driver error, timeout) or a stored record cannot be read (for diff --git a/src/AgentExperience.Abstractions/LifecycleEvent.cs b/src/AgentExperience.Abstractions/LifecycleEvent.cs index b7a6c01..2eba2f0 100644 --- a/src/AgentExperience.Abstractions/LifecycleEvent.cs +++ b/src/AgentExperience.Abstractions/LifecycleEvent.cs @@ -1,9 +1,9 @@ namespace AgentExperience.Abstractions; /// -/// The canonical lifecycle states an Experience Record can occupy. Ownership of the state -/// machine and valid transitions belongs to a later story; this package fixes only the shape of -/// the enum and the event that carries transitions between its values. +/// The canonical lifecycle states an Experience Record can occupy. This package fixes only the shape +/// of the enum and the event that carries transitions between its values; ownership of the state +/// machine and which transitions are valid belongs to Core's lifecycle service. /// public enum ExperienceStatus { @@ -35,8 +35,9 @@ public enum ExperienceStatus /// /// An append-only record of a single lifecycle state transition for an Experience Record. /// Lifecycle changes are events first; current state is a projection derived from them. This -/// package defines only the event's data shape — transition validity rules belong to a later -/// story. +/// package defines only the event's data shape. Which transitions are valid is owned by Core's +/// lifecycle service, and the store enforces that decision against real state by matching +/// and when it applies the event. /// /// Unique identifier for this lifecycle event. /// The Experience Record this event applies to. diff --git a/src/AgentExperience.Core/Lifecycle/ExperienceLifecycleService.cs b/src/AgentExperience.Core/Lifecycle/ExperienceLifecycleService.cs new file mode 100644 index 0000000..227e19c --- /dev/null +++ b/src/AgentExperience.Core/Lifecycle/ExperienceLifecycleService.cs @@ -0,0 +1,141 @@ +using AgentExperience.Abstractions; + +namespace AgentExperience.Core.Lifecycle; + +/// +/// Core's lifecycle owner: it decides whether a requested transition is legal, stamps the +/// that records it, and hands that event to the +/// port to append atomically with the record's projection. The +/// adapter persists the decision as given; it never invents a transition, a status, or a score +/// (ARCHITECTURE-SPINE AD-6). +/// +/// +/// +/// This version allows only the minimal table the first durable lifecycle needs: +/// to ; any status +/// other than to ; +/// and any status to . Anything else is refused here, with +/// , and never reaches the store. +/// Reinforcement, contest, staleness, and supersession are later stories. The table is a Core decision, +/// but it is not Core's only defence: the store matches the event's +/// against the record's real status, so a caller that asserts a +/// prior status the record is not in gets rather +/// than a committed forbidden transition. +/// +/// +/// Reading a record's history is deliberately not mirrored here. It is a plain scoped read with no +/// lifecycle decision in it, so it stays on the +/// port rather than becoming a pass-through this service would only forward. +/// +/// +/// Nothing here computes reuse confidence or counters, decides storage or risk policy, or orchestrates +/// finalization. A store outcome is surfaced one-to-one, so a database failure can never be reported as +/// a durable success: infrastructure failures throw and caller +/// cancellation surfaces as an unwrapped , both straight from +/// the port. +/// +/// +public sealed class ExperienceLifecycleService +{ + private static readonly IReadOnlyList NoErrors = []; + + private readonly IExperienceRecordStore _store; + + /// Creates a lifecycle service over a record store. + /// The port that persists events and projections atomically. + /// is . + public ExperienceLifecycleService(IExperienceRecordStore store) + { + ArgumentNullException.ThrowIfNull(store); + _store = store; + } + + /// + /// Determines whether this version of the lifecycle allows moving a record from + /// to . An undefined enum value is + /// never allowed, so an external caller cannot read this as permission to attempt one. (Inside + /// an undefined value is instead passed through to the store, which + /// reports it as with a field path, so a malformed + /// request is not reported as a policy refusal.) + /// + /// The status the transition starts from. + /// The status the transition moves to. + /// when the transition is in the allowed table. + public static bool IsTransitionAllowed(ExperienceStatus priorStatus, ExperienceStatus currentStatus) => + Enum.IsDefined(priorStatus) + && Enum.IsDefined(currentStatus) + && ((priorStatus == ExperienceStatus.Candidate && currentStatus == ExperienceStatus.Validated) + || (currentStatus == ExperienceStatus.Quarantined && priorStatus != ExperienceStatus.Revoked) + || currentStatus == ExperienceStatus.Revoked); + + /// + /// Validates the requested transition, stamps its , and commits it + /// through the store. + /// + /// What the host has established the caller may do. Passed to the store unchanged. + /// The transition to commit. + /// Cancels the operation. + /// The store's outcome, surfaced unchanged, or when Core refused before calling it. + /// or is . + /// Storage infrastructure failed. Lifecycle state is unchanged. + /// was cancelled. + public async Task CommitAsync( + AuthorizationContext authorization, + CommitLifecycleTransitionRequest request, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(authorization); + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(request.Scope, $"{nameof(request)}.{nameof(request.Scope)}"); + + // A null PriorStatus is a record's first event: there is no starting status to look up, and the + // store skips its status match too. Only a well-formed pair can be looked up in the table at + // all; an undefined enum value is a malformed request, which the store reports with its field path. + if (request.PriorStatus is { } priorStatus + && Enum.IsDefined(priorStatus) + && Enum.IsDefined(request.CurrentStatus) + && !IsTransitionAllowed(priorStatus, request.CurrentStatus)) + { + return new( + LifecycleTransitionOutcome.TransitionNotAllowed, + Event: null, + Revision: 0, + CurrentStatus: null, + NoErrors, + $"Moving a record from {priorStatus} to {request.CurrentStatus} is not an allowed transition."); + } + + var lifecycleEvent = new LifecycleEvent( + EventId: request.EventId, + ExperienceRecordId: request.ExperienceId, + PriorStatus: request.PriorStatus, + CurrentStatus: request.CurrentStatus, + Reason: request.Reason, + Producer: request.Producer, + OccurredAt: request.OccurredAt, + ExpectedRevision: request.ExpectedRevision); + + var result = await _store + .CommitLifecycleEventAsync(authorization, request.Scope, lifecycleEvent, cancellationToken) + .ConfigureAwait(false); + + return new(ToTransitionOutcome(result.Outcome), lifecycleEvent, result.Revision, result.CurrentStatus, result.Errors, Reason: null); + } + + /// + /// Maps a store outcome to its lifecycle counterpart one-to-one. An outcome this operation cannot + /// produce is a contract violation by the store, not something to silently reinterpret. + /// + private static LifecycleTransitionOutcome ToTransitionOutcome(ExperienceStoreOutcome outcome) => outcome switch + { + ExperienceStoreOutcome.Committed => LifecycleTransitionOutcome.Committed, + ExperienceStoreOutcome.StaleRevision => LifecycleTransitionOutcome.StaleRevision, + ExperienceStoreOutcome.StatusMismatch => LifecycleTransitionOutcome.StatusMismatch, + ExperienceStoreOutcome.Conflict => LifecycleTransitionOutcome.Conflict, + ExperienceStoreOutcome.NotFound => LifecycleTransitionOutcome.NotFound, + ExperienceStoreOutcome.Denied => LifecycleTransitionOutcome.Denied, + ExperienceStoreOutcome.Invalid => LifecycleTransitionOutcome.Invalid, + _ => throw new ExperienceStoreException( + $"The Experience Record store returned '{outcome}', which is not a lifecycle commit outcome."), + }; +} diff --git a/src/AgentExperience.Core/Lifecycle/LifecycleResults.cs b/src/AgentExperience.Core/Lifecycle/LifecycleResults.cs new file mode 100644 index 0000000..0f18d5f --- /dev/null +++ b/src/AgentExperience.Core/Lifecycle/LifecycleResults.cs @@ -0,0 +1,110 @@ +using AgentExperience.Abstractions; + +namespace AgentExperience.Core.Lifecycle; + +/// +/// One requested lifecycle transition, submitted to +/// . Every identifying value is caller-supplied so +/// a retry is byte-for-byte the same request: is the idempotency key the store +/// deduplicates on, and is part of the event's stored identity, so neither may +/// be regenerated on a retry. +/// +/// Unique identifier for this transition, and this call's idempotency key. Must not be . +/// The record to transition. Must not be . +/// The exact scope the record lies in. Never treated as authority. +/// +/// The record's status this transition starts from, as the caller read it. Checked against the +/// allowed-transition table, stamped onto the event, and matched against the record's stored status when +/// the store applies it -- so asserting the wrong one is a +/// , never a committed forbidden transition. +/// records a first event on a record that has no lifecycle history yet, which +/// skips both the transition table and the store's status match. +/// +/// The status to move the record to. +/// Auditable, human-readable reason for the transition. Never private reasoning. Must be non-blank. +/// Identity of whatever produced this transition (a policy, an evaluator, or a human principal identifier). Must be non-blank. +/// When this transition was decided. +/// The record's revision this transition was decided against. Must equal the stored revision or the commit is refused as stale. +public sealed record CommitLifecycleTransitionRequest( + Guid EventId, + Guid ExperienceId, + Scope Scope, + ExperienceStatus? PriorStatus, + ExperienceStatus CurrentStatus, + string Reason, + string Producer, + DateTimeOffset OccurredAt, + long ExpectedRevision); + +/// +/// The disposition a call reached. Every member +/// except is the store port's own +/// , surfaced one-to-one and never reinterpreted; +/// is the one decision Core makes on its own, before the port is +/// called at all. +/// +public enum LifecycleTransitionOutcome +{ + /// + /// The event was appended and the record's projection updated in one transaction, or an identical + /// replay reported the original commit. Revision is the record's revision after it. + /// + Committed, + + /// + /// Core refused: the requested to + /// move is not one this version + /// allows. No store call was made and nothing was written. + /// + TransitionNotAllowed, + + /// + /// The record's revision had already moved past + /// . Nothing was written; + /// Revision carries the record's current revision to re-decide against. + /// + StaleRevision, + + /// + /// The record was not in the the + /// transition was decided against, so the store refused it even though the revision matched. Nothing + /// was written; CurrentStatus carries the status the record is actually in. + /// + StatusMismatch, + + /// + /// This is already stored with at least one + /// differing field. Nothing was written and the stored event is unchanged. + /// + Conflict, + + /// No record with that ID exists within the requested scope (including when it exists in another scope). + NotFound, + + /// The request scope lies outside the host-established authorization. No storage was accessed. + Denied, + + /// The request was malformed. See Errors. No storage was accessed. + Invalid, +} + +/// +/// The result of one call. +/// +/// What happened. +/// +/// The event Core stamped and issued to the store, when the request passed the transition table; +/// on . Present even +/// when the store wrote nothing, so a caller can log exactly what was attempted. +/// +/// The record's revision after the commit, or its current revision on and ; otherwise 0. +/// The record's stored status on , to re-decide the transition against; otherwise . +/// Every store validation error when is ; otherwise empty. +/// Optional, auditable, content-free explanation, e.g. why Core refused the transition. +public sealed record CommitLifecycleTransitionResult( + LifecycleTransitionOutcome Outcome, + LifecycleEvent? Event, + long Revision, + ExperienceStatus? CurrentStatus, + IReadOnlyList Errors, + string? Reason); diff --git a/src/AgentExperience.Storage.Postgres/AgentExperience.Storage.Postgres.csproj b/src/AgentExperience.Storage.Postgres/AgentExperience.Storage.Postgres.csproj index 5e2c110..f024a5d 100644 --- a/src/AgentExperience.Storage.Postgres/AgentExperience.Storage.Postgres.csproj +++ b/src/AgentExperience.Storage.Postgres/AgentExperience.Storage.Postgres.csproj @@ -23,6 +23,7 @@ + diff --git a/src/AgentExperience.Storage.Postgres/ExperienceRecordValidator.cs b/src/AgentExperience.Storage.Postgres/ExperienceRecordValidator.cs index 1a68d5d..184ecb7 100644 --- a/src/AgentExperience.Storage.Postgres/ExperienceRecordValidator.cs +++ b/src/AgentExperience.Storage.Postgres/ExperienceRecordValidator.cs @@ -72,6 +72,58 @@ public static IReadOnlyList ValidateGet(Scope scope, Guid return errors; } + /// + /// Validates a lifecycle commit: the event's own fields plus the request scope the record must lie + /// in. Field paths name the member, so a caller can map an error back + /// to what it supplied. + /// + public static IReadOnlyList ValidateLifecycleEvent(Scope scope, LifecycleEvent lifecycleEvent) + { + var errors = new List(); + + if (lifecycleEvent.EventId == Guid.Empty) + { + errors.Add(new("EventId", "must not be an empty GUID.")); + } + + if (lifecycleEvent.ExperienceRecordId == Guid.Empty) + { + errors.Add(new("ExperienceRecordId", "must not be an empty GUID.")); + } + + ValidateScope(scope, "Scope", errors); + + if (lifecycleEvent.PriorStatus is { } priorStatus) + { + RequireDefined(priorStatus, "PriorStatus", errors); + } + + RequireDefined(lifecycleEvent.CurrentStatus, "CurrentStatus", errors); + RequireNotBlank(lifecycleEvent.Reason, "Reason", errors); + RequireNotBlank(lifecycleEvent.Producer, "Producer", errors); + + if (lifecycleEvent.OccurredAt == default) + { + // OccurredAt is part of the event's stored identity, so an unset value would silently become + // part of the idempotency key. No upper bound: clock skew makes a future check unsafe. + errors.Add(new("OccurredAt", "must be set to when the transition occurred.")); + } + + if (lifecycleEvent.ExpectedRevision < 0) + { + errors.Add(new("ExpectedRevision", "must not be negative.")); + } + else if (lifecycleEvent.ExpectedRevision >= long.MaxValue - 1) + { + // A successful commit stores ExpectedRevision + 1, which would wrap silently at long.MaxValue. + // One below it is rejected too: committing there would leave the record permanently stuck, + // because every later commit would expect an unrepresentable revision. + errors.Add(new("ExpectedRevision", "must leave room for the next revision, so a later commit stays possible.")); + } + + return errors; + } + public static IReadOnlyList ValidateQuery(ExperienceRecordQuery query) { var errors = new List(); diff --git a/src/AgentExperience.Storage.Postgres/ExperienceSchemaMigrator.cs b/src/AgentExperience.Storage.Postgres/ExperienceSchemaMigrator.cs index 6a38f3a..89b2144 100644 --- a/src/AgentExperience.Storage.Postgres/ExperienceSchemaMigrator.cs +++ b/src/AgentExperience.Storage.Postgres/ExperienceSchemaMigrator.cs @@ -25,8 +25,9 @@ namespace AgentExperience.Storage.Postgres; /// and one for the scripts, and must not be multiplexing, because a multiplexed command does not stay /// on one physical connection and so cannot hold a session advisory lock. The migrating role needs /// CREATE on the database (to create the agent_experience schema) and on that schema (to -/// create its tables). The store itself only needs INSERT and SELECT on -/// agent_experience.experience_records. +/// create its tables). The store itself only needs SELECT, INSERT, and UPDATE on +/// agent_experience.experience_records and SELECT and INSERT on +/// agent_experience.lifecycle_events. /// /// /// The wait for the advisory lock is deliberately unbounded and ends only with the caller's token. Each diff --git a/src/AgentExperience.Storage.Postgres/Migrations/0002_create_lifecycle_events.sql b/src/AgentExperience.Storage.Postgres/Migrations/0002_create_lifecycle_events.sql new file mode 100644 index 0000000..f63da87 --- /dev/null +++ b/src/AgentExperience.Storage.Postgres/Migrations/0002_create_lifecycle_events.sql @@ -0,0 +1,56 @@ +-- AgentExperience.NET: append-only lifecycle event log for Experience Records. +-- Applied by ExperienceSchemaMigrator and recorded in agent_experience.schema_versions. +-- Every statement is IF NOT EXISTS on purpose, matching 0001, so a database whose schema was applied by +-- hand can still be journaled. Do not edit this script once it has been journaled anywhere; add the +-- next-numbered script instead. (This script is still unreleased and has only ever been applied to +-- throwaway test databases, so the unique index below was corrected in place during review; once this +-- branch ships, the append-only rule applies to it as it does to 0001.) +-- +-- Each row is one committed transition. The store writes a row and the matching experience_records +-- projection update in a single transaction, so the log and the projection can never disagree. Rows are +-- never updated or deleted: event_id is the idempotency key a replay is compared against, and +-- applied_revision (always expected_revision + 1) is the record revision this event produced. +-- +-- There is deliberately no foreign key to experience_records: a commit for a record that does not exist +-- in the request scope is rolled back by the store's own revision-checked projection update, and a +-- foreign-key violation would report it as an infrastructure failure instead of a NotFound outcome. + +CREATE TABLE IF NOT EXISTS agent_experience.lifecycle_events ( + event_id uuid NOT NULL, + experience_id uuid NOT NULL, + tenant_id text NOT NULL, + application_id text NOT NULL, + project_id text NOT NULL, + team_id text NULL, + agent_id text NULL, + user_id text NULL, + prior_status text NULL, + current_status text NOT NULL, + reason text NOT NULL, + producer text NOT NULL, + occurred_at timestamptz NOT NULL, + recorded_at timestamptz NOT NULL, + expected_revision bigint NOT NULL, + applied_revision bigint NOT NULL, + CONSTRAINT lifecycle_events_pkey PRIMARY KEY (event_id), + CONSTRAINT lifecycle_events_event_id_not_empty CHECK (event_id <> '00000000-0000-0000-0000-000000000000'::uuid), + CONSTRAINT lifecycle_events_experience_id_not_empty CHECK (experience_id <> '00000000-0000-0000-0000-000000000000'::uuid), + CONSTRAINT lifecycle_events_tenant_id_not_blank CHECK (tenant_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_application_id_not_blank CHECK (application_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_project_id_not_blank CHECK (project_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_team_id_not_blank CHECK (team_id IS NULL OR team_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_agent_id_not_blank CHECK (agent_id IS NULL OR agent_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_user_id_not_blank CHECK (user_id IS NULL OR user_id ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_prior_status_not_blank CHECK (prior_status IS NULL OR prior_status ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_current_status_not_blank CHECK (current_status ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_reason_not_blank CHECK (reason ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_producer_not_blank CHECK (producer ~ '[^[:space:]]'), + CONSTRAINT lifecycle_events_expected_revision_nonnegative CHECK (expected_revision >= 0), + CONSTRAINT lifecycle_events_applied_revision_follows_expected CHECK (applied_revision = expected_revision + 1) +); + +-- Unique, not merely an index: exactly one event may claim a given revision of a record, so the log can +-- never desynchronize from the projection even if a second writer bypasses the store. Two racing commits +-- from the same revision collide here, and the store reports the loser as a stale revision. +CREATE UNIQUE INDEX IF NOT EXISTS ix_lifecycle_events_record_revision + ON agent_experience.lifecycle_events (experience_id, applied_revision); diff --git a/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordSchema.cs b/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordSchema.cs index 893cf14..920fe7e 100644 --- a/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordSchema.cs +++ b/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordSchema.cs @@ -19,10 +19,13 @@ public static class PostgresExperienceRecordSchema /// The initial script that creates the experience_records table. public const string InitialScriptName = "0001_create_experience_records.sql"; + /// The script that creates the append-only lifecycle_events table. + public const string LifecycleEventsScriptName = "0002_create_lifecycle_events.sql"; + private const string ResourcePrefix = "AgentExperience.Storage.Postgres.Migrations."; /// Every embedded script name, in the order they must be applied. - public static IReadOnlyList ScriptNames { get; } = [InitialScriptName]; + public static IReadOnlyList ScriptNames { get; } = [InitialScriptName, LifecycleEventsScriptName]; /// Reads an embedded script's SQL text. /// One of . diff --git a/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordStore.cs b/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordStore.cs index 981d149..000ffaa 100644 --- a/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordStore.cs +++ b/src/AgentExperience.Storage.Postgres/PostgresExperienceRecordStore.cs @@ -15,6 +15,11 @@ namespace AgentExperience.Storage.Postgres; /// never migrates, on construction or otherwise. /// /// +/// is the only operation that changes a stored record: it appends +/// the event and updates the record's projection in one transaction on one connection, keyed by +/// for idempotency and by +/// for concurrency. The store persists the transition Core +/// decided and never derives a status, score, or counter of its own. /// PostgreSQL timestamptz stores microseconds, so and /// are truncated to whole microseconds (in UTC) on write. /// Nested timestamps live in the JSONB payload at full precision and are also returned in UTC. @@ -53,6 +58,61 @@ public sealed class PostgresExperienceRecordStore : IExperienceRecordStore private const string QueryOrderAndLimit = " ORDER BY created_at DESC, experience_id LIMIT @limit"; + private const string EventsTable = "agent_experience.lifecycle_events"; + + private const string EventColumns = + "event_id, experience_id, tenant_id, application_id, project_id, team_id, agent_id, user_id, " + + "prior_status, current_status, reason, producer, occurred_at, recorded_at, expected_revision, applied_revision"; + + private const string InsertEventSql = + $"INSERT INTO {EventsTable} ({EventColumns}) VALUES (@event_id, @experience_id, @tenant_id, @application_id, " + + "@project_id, @team_id, @agent_id, @user_id, @prior_status, @current_status, @reason, @producer, " + + "@occurred_at, @recorded_at, @expected_revision, @applied_revision)"; + + /// The primary key a resubmitted violates. + private const string EventPrimaryKey = "lifecycle_events_pkey"; + + /// The unique index a second event claiming an already-taken record revision violates. + private const string EventRevisionIndex = "ix_lifecycle_events_record_revision"; + + /// + /// The revision guard, the prior-status guard, and the scope predicate live in the same statement, + /// so a stale revision, a prior status the record is not in, and a foreign scope are all "no row + /// updated" and none of them can overwrite state it does not own. A + /// @prior_status (a record's first event) skips the status match. + /// + private const string UpdateProjectionSql = + $"UPDATE {Table} SET status = @current_status, revision = @applied_revision, updated_at = @recorded_at " + + $"WHERE experience_id = @experience_id AND revision = @expected_revision " + + $"AND (@prior_status IS NULL OR status = @prior_status) AND {ScopePredicate}"; + + private const string SelectRevisionAndStatusSql = + $"SELECT revision, status FROM {Table} WHERE experience_id = @experience_id AND {ScopePredicate}"; + + private const string SelectEventSql = $"SELECT {EventColumns} FROM {EventsTable} WHERE event_id = @event_id"; + + private const string JoinedEventColumns = + "e.event_id, e.experience_id, e.tenant_id, e.application_id, e.project_id, e.team_id, e.agent_id, e.user_id, " + + "e.prior_status, e.current_status, e.reason, e.producer, e.occurred_at, e.recorded_at, e.expected_revision, " + + "e.applied_revision"; + + private const string RecordScopePredicate = + "r.tenant_id = @tenant_id AND r.application_id = @application_id AND r.project_id = @project_id " + + "AND r.team_id IS NOT DISTINCT FROM @team_id AND r.agent_id IS NOT DISTINCT FROM @agent_id " + + "AND r.user_id IS NOT DISTINCT FROM @user_id"; + + /// + /// One statement, so the revision and the events come from one snapshot however the server is + /// configured: a commit landing mid-read can never make the returned revision contradict the + /// returned events. The outer join keeps a record with no events a + /// with an empty history -- that row has a null event_id. + /// + private const string HistorySql = + $"SELECT {JoinedEventColumns}, r.revision FROM {Table} r " + + $"LEFT JOIN {EventsTable} e ON e.experience_id = r.experience_id " + + $"WHERE r.experience_id = @experience_id AND {RecordScopePredicate} " + + "ORDER BY e.applied_revision"; + private static readonly IReadOnlyList NoErrors = []; private readonly NpgsqlDataSource _dataSource; @@ -237,6 +297,284 @@ public async Task QueryAsync( } } + /// + public async Task CommitLifecycleEventAsync( + AuthorizationContext authorization, + Scope scope, + LifecycleEvent lifecycleEvent, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(authorization); + ArgumentNullException.ThrowIfNull(scope); + ArgumentNullException.ThrowIfNull(lifecycleEvent); + + var errors = ExperienceRecordValidator.ValidateLifecycleEvent(scope, lifecycleEvent); + if (errors.Count > 0) + { + return new(ExperienceStoreOutcome.Invalid, 0, null, errors); + } + + if (!authorization.Permits(scope)) + { + return new(ExperienceStoreOutcome.Denied, 0, null, NoErrors); + } + + cancellationToken.ThrowIfCancellationRequested(); + + // Both timestamps are truncated the same way the record's columns are, so a replay's stored + // OccurredAt compares equal to the value the caller resubmits. + var occurredAt = ToStoredTimestamp(lifecycleEvent.OccurredAt); + var recordedAt = ToStoredTimestamp(DateTimeOffset.UtcNow); + var appliedRevision = lifecycleEvent.ExpectedRevision + 1; + + try + { + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); + + // Pinned, not inherited: under REPEATABLE READ or SERIALIZABLE the same-revision race would + // abort with a serialization failure instead of matching no row, turning an expected stale + // revision into an infrastructure failure. + await using var transaction = await connection + .BeginTransactionAsync(System.Data.IsolationLevel.ReadCommitted, cancellationToken).ConfigureAwait(false); + + try + { + await using var insert = new NpgsqlCommand(InsertEventSql, connection, transaction); + AddEventParameters(insert.Parameters, scope, lifecycleEvent, occurredAt, recordedAt, appliedRevision); + await insert.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + catch (PostgresException ex) when (IsViolationOf(ex, EventPrimaryKey, cancellationToken)) + { + // A resubmitted event ID. PostgreSQL has aborted the transaction, so nothing this call + // attempted survives; the stored row then decides replay from conflict. + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + return await CompareStoredEventAsync(connection, scope, lifecycleEvent, occurredAt, cancellationToken).ConfigureAwait(false); + } + catch (PostgresException ex) when (IsViolationOf(ex, EventRevisionIndex, cancellationToken)) + { + // A different event already claimed this record revision. The unique index makes the + // loser of a same-revision race block here and fail once the winner commits, which is a + // stale revision by another name. + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + return await StaleOrMissingAsync(connection, null, scope, lifecycleEvent.ExperienceRecordId, cancellationToken).ConfigureAwait(false); + } + + int updated; + try + { + await using var update = new NpgsqlCommand(UpdateProjectionSql, connection, transaction); + var parameters = update.Parameters; + parameters.Add(new NpgsqlParameter("experience_id", lifecycleEvent.ExperienceRecordId)); + parameters.Add(new NpgsqlParameter("current_status", lifecycleEvent.CurrentStatus.ToString())); + parameters.Add(NullableText("prior_status", lifecycleEvent.PriorStatus?.ToString())); + parameters.Add(new NpgsqlParameter("expected_revision", lifecycleEvent.ExpectedRevision)); + parameters.Add(new NpgsqlParameter("applied_revision", appliedRevision)); + parameters.Add(new NpgsqlParameter("recorded_at", recordedAt)); + AddScopeParameters(parameters, scope); + updated = await update.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + catch (PostgresException ex) when (!cancellationToken.IsCancellationRequested + && ex.SqlState is PostgresErrorCodes.SerializationFailure or PostgresErrorCodes.DeadlockDetected) + { + // Another writer got there first. However the server is configured, losing that race is an + // expected condition, not an infrastructure failure. + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + return await StaleOrMissingAsync(connection, null, scope, lifecycleEvent.ExperienceRecordId, cancellationToken).ConfigureAwait(false); + } + + if (updated == 0) + { + // The record is not in this scope, its revision has moved on, or it is not in the status + // the event was decided against. The row is re-read inside the same transaction that is + // about to be rolled back, so the event insert above never reaches the log. + var current = await ReadRevisionAndStatusAsync(connection, transaction, scope, lifecycleEvent.ExperienceRecordId, cancellationToken) + .ConfigureAwait(false); + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + + if (current is not { } record) + { + return new(ExperienceStoreOutcome.NotFound, 0, null, NoErrors); + } + + return record.Revision != lifecycleEvent.ExpectedRevision + ? new(ExperienceStoreOutcome.StaleRevision, record.Revision, null, NoErrors) + // Scope and revision both matched, so the prior-status guard is what rejected it. + : new(ExperienceStoreOutcome.StatusMismatch, record.Revision, record.Status, NoErrors); + } + + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + return new(ExperienceStoreOutcome.Committed, appliedRevision, null, NoErrors); + } + catch (Exception ex) when (IsInfrastructureFailure(ex, cancellationToken)) + { + // Nothing this call wrote is visible unless the commit itself succeeded and only its + // acknowledgement was lost; retrying the identical event then replays instead of reapplying. + throw Translate(ex, "lifecycle commit", cancellationToken); + } + } + + /// + public async Task GetHistoryAsync( + AuthorizationContext authorization, + Scope scope, + Guid experienceId, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(authorization); + ArgumentNullException.ThrowIfNull(scope); + + var errors = ExperienceRecordValidator.ValidateGet(scope, experienceId); + if (errors.Count > 0) + { + return new(ExperienceStoreOutcome.Invalid, 0, [], errors); + } + + if (!authorization.Permits(scope)) + { + return new(ExperienceStoreOutcome.Denied, 0, [], NoErrors); + } + + cancellationToken.ThrowIfCancellationRequested(); + + try + { + await using var command = _dataSource.CreateCommand(HistorySql); + command.Parameters.Add(new NpgsqlParameter("experience_id", experienceId)); + AddScopeParameters(command.Parameters, scope); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + // No record in this scope: indistinguishable from one that exists elsewhere. + return new(ExperienceStoreOutcome.NotFound, 0, [], NoErrors); + } + + var revision = ReadRevision(reader, 16); + + var events = new List(); + if (!reader.IsDBNull(0)) + { + // A null event_id is the outer join's single "record with no events" row. + do + { + events.Add(ReadEvent(reader)); + } + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)); + } + + return new(ExperienceStoreOutcome.Found, revision, events, NoErrors); + } + catch (Exception ex) when (IsInfrastructureFailure(ex, cancellationToken)) + { + throw Translate(ex, "history", cancellationToken); + } + } + + /// + /// Decides a resubmitted : byte-for-byte the same event (scope + /// included) is the original commit replayed, so its original outcome is returned and nothing is + /// written; any difference is a . The comparison is + /// identical whichever scope owns the stored event, so it reveals no event data. + /// + private static async Task CompareStoredEventAsync( + NpgsqlConnection connection, + Scope scope, + LifecycleEvent lifecycleEvent, + DateTimeOffset occurredAt, + CancellationToken cancellationToken) + { + await using var command = new NpgsqlCommand(SelectEventSql, connection); + command.Parameters.Add(new NpgsqlParameter("event_id", lifecycleEvent.EventId)); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + // Events are never deleted, so the row that just collided cannot vanish. Treat the + // impossible case as a conflict rather than writing anything. + return new(ExperienceStoreOutcome.Conflict, 0, null, NoErrors); + } + + var stored = ReadEvent(reader); + var storedScope = ReadEventScope(reader); + var appliedRevision = ReadRevision(reader, 15); + + // Record equality compares every field of the event; the scope is compared alongside it. The + // revision reported is the one the original commit produced, not the record's current one. + var resubmitted = lifecycleEvent with { OccurredAt = occurredAt }; + return stored == resubmitted && storedScope == scope + ? new(ExperienceStoreOutcome.Committed, appliedRevision, null, NoErrors) + : new(ExperienceStoreOutcome.Conflict, 0, null, NoErrors); + } + + /// + /// Reports a lost race: the record's current revision, or + /// when it is not in this scope at all. Used where the server aborted the transaction itself, so the + /// re-read runs outside it. + /// + private static async Task StaleOrMissingAsync( + NpgsqlConnection connection, + NpgsqlTransaction? transaction, + Scope scope, + Guid experienceId, + CancellationToken cancellationToken) + { + var current = await ReadRevisionAndStatusAsync(connection, transaction, scope, experienceId, cancellationToken).ConfigureAwait(false); + return current is { } record + ? new(ExperienceStoreOutcome.StaleRevision, record.Revision, null, NoErrors) + : new(ExperienceStoreOutcome.NotFound, 0, null, NoErrors); + } + + private static async Task<(long Revision, ExperienceStatus Status)?> ReadRevisionAndStatusAsync( + NpgsqlConnection connection, + NpgsqlTransaction? transaction, + Scope scope, + Guid experienceId, + CancellationToken cancellationToken) + { + await using var command = new NpgsqlCommand(SelectRevisionAndStatusSql, connection, transaction); + command.Parameters.Add(new NpgsqlParameter("experience_id", experienceId)); + AddScopeParameters(command.Parameters, scope); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return null; + } + + return (ReadRevision(reader, 0), ReadStoredStatus(reader, 1)); + } + + /// + /// Matches a unique violation of one named constraint. Naming it keeps the event primary key (a + /// resubmitted event ID) apart from the record-revision index (a lost race), so neither is ever + /// mistaken for the other or for a constraint added later. + /// + private static bool IsViolationOf(PostgresException ex, string constraintName, CancellationToken cancellationToken) => + ex.SqlState == PostgresErrorCodes.UniqueViolation + && string.Equals(ex.ConstraintName, constraintName, StringComparison.Ordinal) + && !cancellationToken.IsCancellationRequested; + + private static void AddEventParameters( + NpgsqlParameterCollection parameters, + Scope scope, + LifecycleEvent lifecycleEvent, + DateTimeOffset occurredAt, + DateTimeOffset recordedAt, + long appliedRevision) + { + parameters.Add(new NpgsqlParameter("event_id", lifecycleEvent.EventId)); + parameters.Add(new NpgsqlParameter("experience_id", lifecycleEvent.ExperienceRecordId)); + AddScopeParameters(parameters, scope); + parameters.Add(NullableText("prior_status", lifecycleEvent.PriorStatus?.ToString())); + parameters.Add(new NpgsqlParameter("current_status", lifecycleEvent.CurrentStatus.ToString())); + parameters.Add(new NpgsqlParameter("reason", lifecycleEvent.Reason)); + parameters.Add(new NpgsqlParameter("producer", lifecycleEvent.Producer)); + parameters.Add(new NpgsqlParameter("occurred_at", occurredAt)); + parameters.Add(new NpgsqlParameter("recorded_at", recordedAt)); + parameters.Add(new NpgsqlParameter("expected_revision", lifecycleEvent.ExpectedRevision)); + parameters.Add(new NpgsqlParameter("applied_revision", appliedRevision)); + } + private static void AddScopeParameters(NpgsqlParameterCollection parameters, Scope scope) { parameters.Add(new NpgsqlParameter("tenant_id", NpgsqlDbType.Text) { TypedValue = scope.TenantId }); @@ -292,15 +630,79 @@ private static ExperienceRecord ReadRecord(DbDataReader reader) } } - private static ExperienceRecord DecodeRecord(DbDataReader reader) + private static LifecycleEvent ReadEvent(DbDataReader reader) + { + try + { + return DecodeEvent(reader); + } + catch (Exception ex) when (ex is not (ExperienceStoreException or OperationCanceledException or NpgsqlException)) + { + // Schema drift or a corrupt row (e.g. InvalidCastException on a retyped column). + throw new ExperienceStoreException("Stored lifecycle event could not be decoded.", ex); + } + } + + private static LifecycleEvent DecodeEvent(DbDataReader reader) => new( + EventId: reader.GetGuid(0), + ExperienceRecordId: reader.GetGuid(1), + PriorStatus: reader.IsDBNull(8) ? null : DecodeStatus(reader.GetString(8), "lifecycle event"), + CurrentStatus: DecodeStatus(reader.GetString(9), "lifecycle event"), + Reason: reader.GetString(10), + Producer: reader.GetString(11), + OccurredAt: reader.GetFieldValue(12), + ExpectedRevision: reader.GetInt64(14)); + + /// Reads a bigint revision, reporting schema drift the way the row decoders do. + private static long ReadRevision(DbDataReader reader, int ordinal) + { + try + { + return reader.GetInt64(ordinal); + } + catch (Exception ex) when (ex is not (ExperienceStoreException or OperationCanceledException or NpgsqlException)) + { + throw new ExperienceStoreException("Stored Experience Record could not be decoded.", ex); + } + } + + private static ExperienceStatus ReadStoredStatus(DbDataReader reader, int ordinal) + { + try + { + return DecodeStatus(reader.GetString(ordinal), "Experience Record"); + } + catch (Exception ex) when (ex is not (ExperienceStoreException or OperationCanceledException or NpgsqlException)) + { + throw new ExperienceStoreException("Stored Experience Record could not be decoded.", ex); + } + } + + private static Scope ReadEventScope(DbDataReader reader) => new( + reader.GetString(2), + reader.GetString(3), + reader.GetString(4), + reader.IsDBNull(5) ? null : reader.GetString(5), + reader.IsDBNull(6) ? null : reader.GetString(6), + reader.IsDBNull(7) ? null : reader.GetString(7)); + + /// The stored status text. + /// Which stored object the text came from, so a failure names the right row. + private static ExperienceStatus DecodeStatus(string statusText, string objectKind) { - var statusText = reader.GetString(9); if (!Enum.TryParse(statusText, ignoreCase: false, out var status) || !Enum.IsDefined(status) || !string.Equals(status.ToString(), statusText, StringComparison.Ordinal)) { - throw new ExperienceStoreException("Stored Experience Record has an unrecognized status."); + throw new ExperienceStoreException($"Stored {objectKind} has an unrecognized status."); } + return status; + } + + private static ExperienceRecord DecodeRecord(DbDataReader reader) + { + var status = DecodeStatus(reader.GetString(9), "Experience Record"); + var payload = ExperiencePayload.Deserialize(reader.GetInt32(16), reader.GetString(17)); var scope = new Scope( diff --git a/src/AgentExperience.Storage.Postgres/README.md b/src/AgentExperience.Storage.Postgres/README.md index 9a777be..2227c5a 100644 --- a/src/AgentExperience.Storage.Postgres/README.md +++ b/src/AgentExperience.Storage.Postgres/README.md @@ -35,6 +35,25 @@ var page = await store.QueryAsync( authorization, new ExperienceRecordQuery(record.Scope, Statuses: [ExperienceStatus.Validated], Limit: 20), cancellationToken); + +// A lifecycle change: the event and the record's projection commit together, or neither does. +var commit = await store.CommitLifecycleEventAsync( + authorization, + record.Scope, + new LifecycleEvent( + EventId: eventId, // this call's idempotency key -- stable across retries + ExperienceRecordId: record.ExperienceId, + PriorStatus: ExperienceStatus.Candidate, + CurrentStatus: ExperienceStatus.Validated, // decided by Core, never by this adapter + Reason: "required checks passed", + Producer: "finalization", + OccurredAt: decidedAt, + ExpectedRevision: record.Revision), // must equal the stored revision + cancellationToken); +// commit.Revision is record.Revision + 1 when commit.Outcome is Committed. + +var history = await store.GetHistoryAsync(authorization, record.Scope, record.ExperienceId, cancellationToken); +// history.Events is every transition, oldest first; history.Revision is the record's current revision. ``` The store never disposes the data source. The host owns it. @@ -62,6 +81,9 @@ The store never disposes the data source. The host owns it. | Scope outside the authorization context | `Denied` (no connection opened) | | Malformed request | `Invalid` with every `StoreValidationError(Path, Message)` (no connection opened) | | ID already exists in any scope | `Conflict` (stored row unchanged) | +| Lifecycle event and projection committed together | `Committed` | +| Lifecycle `ExpectedRevision` ≠ the record's current `Revision` | `StaleRevision` (nothing written) | +| Lifecycle `PriorStatus` ≠ the record's stored `Status` | `StatusMismatch` with the stored status (nothing written) | | Database or driver failure (`NpgsqlException`, `SocketException`, `TimeoutException`) | throws `ExperienceStoreException` with the original as `InnerException` | | Stored row with an unsupported `payload_version` or an unreadable payload | throws `ExperienceStoreException` | | Caller cancellation | throws `OperationCanceledException`, unwrapped | @@ -71,10 +93,48 @@ Validation messages never contain record payload content, and the store does not A create whose acknowledgement was lost (cancelled or timed out after PostgreSQL committed it) returns `Conflict` when retried. After a `Conflict`, call `GetAsync` in your own scope to check whether the stored record is yours. +## Lifecycle commits + +`CommitLifecycleEventAsync` is the only way a stored record's status changes. It appends the `LifecycleEvent` to +`lifecycle_events` and updates the record's `status`, `revision`, and `updated_at` **in one transaction on one +connection**: both writes commit together, or neither does. A failure between them leaves no event and no +projection change. + +The adapter persists the decision exactly as given. It never derives a status, a reuse confidence, or a counter, and +it never invents a transition the command did not carry — deciding which transitions are legal belongs to Core's +`ExperienceLifecycleService` (ARCHITECTURE-SPINE AD-6). Authorization is checked against the request scope before +the transaction opens, exactly as for the store's other operations, and the scope predicate is applied in SQL. + +- **The revision rule.** `ExpectedRevision` must equal the record's current `Revision`. A successful commit sets + the revision to `ExpectedRevision + 1` and reports it as `result.Revision`. Any other value is `StaleRevision`, + writes nothing, and reports the record's *current* revision so you can re-decide against it. Two commits racing + from the same revision therefore end with exactly one applied event and one revision increment. +- **Idempotency by `EventId`.** Replaying an event whose stored fields are identical — including its scope and its + microsecond-truncated `OccurredAt` — returns the original outcome (`Committed`, with the revision that commit + produced) and writes nothing. A stored `EventId` with *any* differing field is `Conflict`, whichever scope owns + it, and writes nothing. So `EventId` and `OccurredAt` must be stable across retries; regenerating either turns a + retry into a second transition. +- **The prior-status guard.** When the event's `PriorStatus` is non-null it must also equal the record's stored + `Status`, matched in the same statement as the revision. That is what keeps Core's transition table enforced + against real state rather than against what the caller asserted, and keeps a stored event from recording a prior + status the record never had. A mismatch is `StatusMismatch`, writes nothing, and reports the record's stored + status as `result.CurrentStatus` so you can re-decide against it. A null `PriorStatus` — a record's first event — + skips the status match. +- **Missing or foreign records.** A record that does not exist in the request scope is `NotFound`, indistinguishable + from a missing one, and nothing is written. +- **A lost acknowledgement.** A commit that was cancelled or timed out after PostgreSQL committed it is recovered by + retrying the *identical* event: the replay path reports the original `Committed` and the revision that commit + produced, without applying it twice. This is why `EventId` and `OccurredAt` must be stable across retries — unlike + a create, where a lost acknowledgement surfaces as `Conflict` and has to be resolved with `GetAsync`. +- **History.** `GetHistoryAsync` returns the record's current `Revision` plus every event, oldest first, in a single + statement, so the revision can never contradict the events even if a commit lands mid-read. Events are + append-only: nothing deletes or rewrites them. `GetAsync` and its result are unchanged by this operation. + ## Schema -The schema lives in the embedded script `Migrations/0001_create_experience_records.sql`. It creates the -`agent_experience` schema and the `experience_records` table: +The schema lives in the embedded scripts under `Migrations/`. + +`0001_create_experience_records.sql` creates the `agent_experience` schema and the `experience_records` table: - Scope, task, status, confidence, counter, revision, and timestamp columns, with `CHECK` constraints for non-blank scope and value ranges. @@ -82,6 +142,21 @@ The schema lives in the embedded script `Migrations/0001_create_experience_recor - A `payload_version` column. This adapter owns versioning, so the domain types carry no version field. - An index on `(tenant_id, application_id, project_id)`. +`0002_create_lifecycle_events.sql` adds the append-only `lifecycle_events` table: + +- `event_id` as the primary key (the commit's idempotency key), the record ID, the same scope columns as + `experience_records`, prior/current status, reason, producer, `occurred_at`/`recorded_at`, and + `expected_revision`/`applied_revision`. +- `CHECK` constraints mirroring `0001` (non-empty IDs, non-blank scope, reason and producer, non-negative revision) + plus `applied_revision = expected_revision + 1`, so a row written outside this store cannot desynchronize the log + from the projection. +- A **unique** index on `(experience_id, applied_revision)`, so exactly one event can ever claim a given revision of + a record and the log cannot desynchronize from the projection. Two commits racing from the same revision collide + here; the loser is reported as `StaleRevision`. +- Deliberately no foreign key to `experience_records`: a commit for a record outside the request scope is rolled + back by the revision-checked projection update, and a foreign-key violation would report that expected condition + as an infrastructure failure instead. + ### Applying it Call `ExperienceSchemaMigrator.MigrateAsync` explicitly at startup, before using the store. The store never migrates @@ -102,8 +177,8 @@ var migration = await ExperienceSchemaMigrator.MigrateAsync(dataSource, cancella - **Serialized across processes.** The whole run holds a PostgreSQL session advisory lock on its own connection, so two hosts starting at once cannot apply the same script twice. The lock is always released. - **Permissions.** The migrating role needs `CREATE` on the database (for the `agent_experience` schema) and on that - schema (for its tables). The store itself only needs `INSERT` and `SELECT` on - `agent_experience.experience_records`. + schema (for its tables). The store itself only needs `SELECT`, `INSERT`, and `UPDATE` on + `agent_experience.experience_records` and `SELECT` and `INSERT` on `agent_experience.lifecycle_events`. - **Connections.** The data source must allow at least two concurrent connections: one for the advisory lock and one for the scripts. A multiplexing data source (`NpgsqlDataSourceBuilder.EnableMultiplexing`) cannot hold a session advisory lock, because its commands do not stay on one physical connection, so it is not supported for migration. @@ -136,10 +211,11 @@ definition, and a rename would reapply it. Change the schema by adding the next- ## Data semantics -- **Create-only.** Each create is a single `INSERT`. Updates, deletes, and lifecycle events belong to later stories. -- **UTC timestamps.** Every timestamp is stored and returned in UTC. `CreatedAt` and `UpdatedAt` are columns, and - PostgreSQL keeps microsecond precision, so sub-microsecond ticks are truncated on write. Nested timestamps are - stored in the payload at full precision. +- **One write path per change.** Each create is a single `INSERT`. The only update is a lifecycle commit, which is + always paired with its event in one transaction (see above). Nothing deletes a record or an event. +- **UTC timestamps.** Every timestamp is stored and returned in UTC. `CreatedAt`, `UpdatedAt`, and a lifecycle + event's `OccurredAt` are columns, and PostgreSQL keeps microsecond precision, so sub-microsecond ticks are + truncated on write. Nested timestamps are stored in the payload at full precision. - **Tool-call arguments** are stored as JSON and read back normalized to `string`, `bool`, `long` (integers that fit), `double`, `null`, `Dictionary`, or `List`. Dictionary key order is not preserved. Whole-number doubles (for example `1.0`) are written as JSON integers, so they read back as `long`. Values that cannot be serialized to JSON (for diff --git a/tests/AgentExperience.Abstractions.Tests/ContractShapeTests.cs b/tests/AgentExperience.Abstractions.Tests/ContractShapeTests.cs index 06e918e..97b9c7e 100644 --- a/tests/AgentExperience.Abstractions.Tests/ContractShapeTests.cs +++ b/tests/AgentExperience.Abstractions.Tests/ContractShapeTests.cs @@ -299,7 +299,7 @@ public void Store_port_operations_take_authorization_and_a_required_cancellation { var methods = typeof(IExperienceRecordStore).GetMethods().OrderBy(m => m.Name, StringComparer.Ordinal).ToList(); - Assert.Equal(["CreateAsync", "GetAsync", "QueryAsync"], methods.Select(m => m.Name)); + Assert.Equal(["CommitLifecycleEventAsync", "CreateAsync", "GetAsync", "GetHistoryAsync", "QueryAsync"], methods.Select(m => m.Name)); Assert.All(methods, method => { var parameters = method.GetParameters(); @@ -308,12 +308,77 @@ public void Store_port_operations_take_authorization_and_a_required_cancellation Assert.False(parameters[^1].HasDefaultValue); }); - Assert.Equal(typeof(Task), methods[0].ReturnType); - Assert.Equal([typeof(AuthorizationContext), typeof(ExperienceRecord), typeof(CancellationToken)], methods[0].GetParameters().Select(p => p.ParameterType)); - Assert.Equal(typeof(Task), methods[1].ReturnType); - Assert.Equal([typeof(AuthorizationContext), typeof(Scope), typeof(Guid), typeof(CancellationToken)], methods[1].GetParameters().Select(p => p.ParameterType)); - Assert.Equal(typeof(Task), methods[2].ReturnType); - Assert.Equal([typeof(AuthorizationContext), typeof(ExperienceRecordQuery), typeof(CancellationToken)], methods[2].GetParameters().Select(p => p.ParameterType)); + Assert.Equal(typeof(Task), methods[0].ReturnType); + Assert.Equal([typeof(AuthorizationContext), typeof(Scope), typeof(LifecycleEvent), typeof(CancellationToken)], methods[0].GetParameters().Select(p => p.ParameterType)); + Assert.Equal(typeof(Task), methods[1].ReturnType); + Assert.Equal([typeof(AuthorizationContext), typeof(ExperienceRecord), typeof(CancellationToken)], methods[1].GetParameters().Select(p => p.ParameterType)); + Assert.Equal(typeof(Task), methods[2].ReturnType); + Assert.Equal([typeof(AuthorizationContext), typeof(Scope), typeof(Guid), typeof(CancellationToken)], methods[2].GetParameters().Select(p => p.ParameterType)); + Assert.Equal(typeof(Task), methods[3].ReturnType); + Assert.Equal([typeof(AuthorizationContext), typeof(Scope), typeof(Guid), typeof(CancellationToken)], methods[3].GetParameters().Select(p => p.ParameterType)); + Assert.Equal(typeof(Task), methods[4].ReturnType); + Assert.Equal([typeof(AuthorizationContext), typeof(ExperienceRecordQuery), typeof(CancellationToken)], methods[4].GetParameters().Select(p => p.ParameterType)); + } + + // Story 2.4: the lifecycle commit and history contracts. + [Fact] + public void Lifecycle_commit_and_history_results_carry_a_revision_and_ordered_events() + { + var commit = new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.Committed, 4, null, []); + Assert.Equal(4, commit.Revision); + Assert.Null(commit.CurrentStatus); + Assert.Empty(commit.Errors); + + var stale = new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.StaleRevision, 9, null, []); + Assert.Equal(ExperienceStoreOutcome.StaleRevision, stale.Outcome); + + // A prior-status mismatch reports the status the record is actually in, to re-decide against. + var mismatch = new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.StatusMismatch, 9, ExperienceStatus.Quarantined, []); + Assert.Equal(ExperienceStatus.Quarantined, mismatch.CurrentStatus); + Assert.Equal(9, mismatch.Revision); + + var invalid = new ExperienceLifecycleCommitResult( + ExperienceStoreOutcome.Invalid, 0, null, [new StoreValidationError("Reason", "must not be empty or whitespace.")]); + Assert.Equal("Reason", Assert.Single(invalid.Errors).Path); + + // A commit result never carries record or event payload back to the caller. + Assert.DoesNotContain( + typeof(ExperienceLifecycleCommitResult).GetProperties(), + p => p.PropertyType == typeof(ExperienceRecord) || p.PropertyType == typeof(LifecycleEvent)); + + var first = new LifecycleEvent(Guid.NewGuid(), Guid.NewGuid(), null, ExperienceStatus.Candidate, "captured", "capture", Now, 0); + var second = first with { EventId = Guid.NewGuid(), PriorStatus = ExperienceStatus.Candidate, CurrentStatus = ExperienceStatus.Validated, ExpectedRevision = 1 }; + var history = new ExperienceRecordHistoryResult(ExperienceStoreOutcome.Found, 2, [first, second], []); + + Assert.Equal(2, history.Revision); + Assert.Equal([0L, 1L], history.Events.Select(e => e.ExpectedRevision)); + Assert.Equal(ExperienceStatus.Validated, history.Events[^1].CurrentStatus); + + var notFound = new ExperienceRecordHistoryResult(ExperienceStoreOutcome.NotFound, 0, [], []); + Assert.Empty(notFound.Events); + } + + [Fact] + public void LifecycleEvent_equality_is_field_by_field_so_a_replay_can_be_told_from_a_conflict() + { + // The store's replay check compares the resubmitted event against the stored one by value. + var original = new LifecycleEvent( + Guid.Parse("11111111-1111-1111-1111-111111111111"), + Guid.Parse("22222222-2222-2222-2222-222222222222"), + ExperienceStatus.Candidate, + ExperienceStatus.Validated, + "verified evidence", + "finalization", + Now, + 3); + + Assert.Equal(original, original with { }); + Assert.NotEqual(original, original with { Reason = "verified evidence " }); + Assert.NotEqual(original, original with { Producer = "Finalization" }); + Assert.NotEqual(original, original with { PriorStatus = null }); + Assert.NotEqual(original, original with { CurrentStatus = ExperienceStatus.Quarantined }); + Assert.NotEqual(original, original with { OccurredAt = Now.AddTicks(10) }); + Assert.NotEqual(original, original with { ExpectedRevision = 4 }); } [Fact] @@ -332,7 +397,7 @@ public void Query_defaults_to_all_statuses_and_a_limit_of_50_within_1_to_500() public void Store_outcomes_results_and_exception_have_the_expected_shape() { Assert.Equal( - ["Created", "Found", "NotFound", "Denied", "Invalid", "Conflict"], + ["Created", "Found", "NotFound", "Denied", "Invalid", "Conflict", "Committed", "StaleRevision", "StatusMismatch"], Enum.GetNames()); var error = new StoreValidationError("Scope.TenantId", "must not be empty or whitespace."); diff --git a/tests/AgentExperience.Core.Tests/ExperienceLifecycleServiceTests.cs b/tests/AgentExperience.Core.Tests/ExperienceLifecycleServiceTests.cs new file mode 100644 index 0000000..c520282 --- /dev/null +++ b/tests/AgentExperience.Core.Tests/ExperienceLifecycleServiceTests.cs @@ -0,0 +1,319 @@ +using AgentExperience.Core.Lifecycle; + +namespace AgentExperience.Core.Tests; + +/// +/// Core owns which transitions are legal (ARCHITECTURE-SPINE AD-6). These tests pin the minimal table +/// this version allows -- Candidate to Validated, anything but Revoked to Quarantined, anything to +/// Revoked -- prove a transition outside it never reaches the store, and prove the store's outcome is +/// surfaced one-to-one rather than reinterpreted. +/// +public class ExperienceLifecycleServiceTests +{ + private static readonly DateTimeOffset Now = new(2026, 9, 18, 10, 0, 0, TimeSpan.Zero); + private static readonly Scope TestScope = new("tenant-1", "app-1", "project-1"); + private static readonly AuthorizationContext Authorization = new("tenant-1", "principal", ["experience:write"], Now); + + private static readonly ExperienceStatus[] EveryStatus = Enum.GetValues(); + + private static CommitLifecycleTransitionRequest Request( + ExperienceStatus? prior, + ExperienceStatus current, + long expectedRevision = 0, + Guid? eventId = null) => new( + EventId: eventId ?? Guid.NewGuid(), + ExperienceId: Guid.NewGuid(), + Scope: TestScope, + PriorStatus: prior, + CurrentStatus: current, + Reason: "verified evidence", + Producer: "finalization", + OccurredAt: Now, + ExpectedRevision: expectedRevision); + + [Fact] + public async Task Candidate_to_Validated_is_allowed_and_reaches_the_store() + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(ExperienceStatus.Candidate, ExperienceStatus.Validated, 2), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.Committed, result.Outcome); + Assert.Equal(3, result.Revision); + Assert.Empty(result.Errors); + Assert.Null(result.Reason); + Assert.Single(store.Commits); + } + + [Fact] + public async Task Every_status_except_Revoked_may_be_quarantined() + { + foreach (var prior in EveryStatus.Where(s => s != ExperienceStatus.Revoked)) + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(prior, ExperienceStatus.Quarantined), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.Committed, result.Outcome); + Assert.Equal(prior, Assert.Single(store.Commits).Event.PriorStatus); + } + } + + [Fact] + public async Task Every_status_may_be_revoked() + { + foreach (var prior in EveryStatus) + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(prior, ExperienceStatus.Revoked), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.Committed, result.Outcome); + Assert.Equal(ExperienceStatus.Revoked, Assert.Single(store.Commits).Event.CurrentStatus); + } + } + + [Theory] + [InlineData(ExperienceStatus.Revoked, ExperienceStatus.Quarantined)] // revocation is terminal except for re-revocation + [InlineData(ExperienceStatus.Validated, ExperienceStatus.Candidate)] // no walking a record back to candidate + [InlineData(ExperienceStatus.Candidate, ExperienceStatus.Reinforced)] // reinforcement is Epic 3 + [InlineData(ExperienceStatus.Validated, ExperienceStatus.Contested)] + [InlineData(ExperienceStatus.Validated, ExperienceStatus.Stale)] + [InlineData(ExperienceStatus.Validated, ExperienceStatus.Superseded)] + [InlineData(ExperienceStatus.Quarantined, ExperienceStatus.Validated)] + public async Task A_transition_outside_the_table_is_refused_by_Core_and_never_reaches_the_store( + ExperienceStatus prior, + ExperienceStatus current) + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(prior, current), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.TransitionNotAllowed, result.Outcome); + Assert.Empty(store.Commits); + Assert.Null(result.Event); + Assert.Equal(0, result.Revision); + Assert.Empty(result.Errors); + Assert.False(string.IsNullOrWhiteSpace(result.Reason)); + } + + /// + /// Every allowed (prior, current) pair, written out rather than derived, so this cannot silently + /// agree with a changed implementation. 8 statuses x 8 statuses = 64 pairs; the 16 below are allowed + /// and the other 48 are not. + /// + private static readonly HashSet<(ExperienceStatus Prior, ExperienceStatus Current)> AllowedPairs = + [ + // Candidate -> Validated (the only promotion this version allows). + (ExperienceStatus.Candidate, ExperienceStatus.Validated), + + // Anything except Revoked -> Quarantined. + (ExperienceStatus.Candidate, ExperienceStatus.Quarantined), + (ExperienceStatus.Validated, ExperienceStatus.Quarantined), + (ExperienceStatus.Quarantined, ExperienceStatus.Quarantined), + (ExperienceStatus.Contested, ExperienceStatus.Quarantined), + (ExperienceStatus.Stale, ExperienceStatus.Quarantined), + (ExperienceStatus.Superseded, ExperienceStatus.Quarantined), + (ExperienceStatus.Reinforced, ExperienceStatus.Quarantined), + + // Anything -> Revoked. + (ExperienceStatus.Candidate, ExperienceStatus.Revoked), + (ExperienceStatus.Validated, ExperienceStatus.Revoked), + (ExperienceStatus.Quarantined, ExperienceStatus.Revoked), + (ExperienceStatus.Contested, ExperienceStatus.Revoked), + (ExperienceStatus.Stale, ExperienceStatus.Revoked), + (ExperienceStatus.Superseded, ExperienceStatus.Revoked), + (ExperienceStatus.Revoked, ExperienceStatus.Revoked), + (ExperienceStatus.Reinforced, ExperienceStatus.Revoked), + ]; + + [Fact] + public void The_allowed_table_is_exactly_the_enumerated_pairs() + { + Assert.Equal(16, AllowedPairs.Count); + Assert.Equal(8, EveryStatus.Length); + + foreach (var prior in EveryStatus) + { + foreach (var current in EveryStatus) + { + Assert.Equal( + AllowedPairs.Contains((prior, current)), + ExperienceLifecycleService.IsTransitionAllowed(prior, current)); + } + } + } + + [Fact] + public void An_undefined_status_is_never_an_allowed_transition_for_an_external_caller() + { + Assert.False(ExperienceLifecycleService.IsTransitionAllowed((ExperienceStatus)999, ExperienceStatus.Revoked)); + Assert.False(ExperienceLifecycleService.IsTransitionAllowed(ExperienceStatus.Candidate, (ExperienceStatus)999)); + Assert.False(ExperienceLifecycleService.IsTransitionAllowed((ExperienceStatus)998, (ExperienceStatus)999)); + } + + [Fact] + public async Task The_stamped_event_carries_the_request_verbatim() + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + var request = Request(ExperienceStatus.Candidate, ExperienceStatus.Validated, 7); + + var result = await service.CommitAsync(Authorization, request, CancellationToken.None); + + var (scope, stamped) = Assert.Single(store.Commits); + Assert.Same(request.Scope, scope); + Assert.Equal(request.EventId, stamped.EventId); + Assert.Equal(request.ExperienceId, stamped.ExperienceRecordId); + Assert.Equal(request.PriorStatus, stamped.PriorStatus); + Assert.Equal(request.CurrentStatus, stamped.CurrentStatus); + Assert.Equal(request.Reason, stamped.Reason); + Assert.Equal(request.Producer, stamped.Producer); + Assert.Equal(request.OccurredAt, stamped.OccurredAt); + Assert.Equal(request.ExpectedRevision, stamped.ExpectedRevision); + Assert.Equal(stamped, result.Event); + + // Nothing is invented: the service never touches confidence, counters, or timestamps of its own. + var replay = await service.CommitAsync(Authorization, request, CancellationToken.None); + Assert.Equal(stamped, replay.Event); + } + + [Theory] + [InlineData(ExperienceStoreOutcome.Committed, LifecycleTransitionOutcome.Committed)] + [InlineData(ExperienceStoreOutcome.StaleRevision, LifecycleTransitionOutcome.StaleRevision)] + [InlineData(ExperienceStoreOutcome.StatusMismatch, LifecycleTransitionOutcome.StatusMismatch)] + [InlineData(ExperienceStoreOutcome.Conflict, LifecycleTransitionOutcome.Conflict)] + [InlineData(ExperienceStoreOutcome.NotFound, LifecycleTransitionOutcome.NotFound)] + [InlineData(ExperienceStoreOutcome.Denied, LifecycleTransitionOutcome.Denied)] + [InlineData(ExperienceStoreOutcome.Invalid, LifecycleTransitionOutcome.Invalid)] + public async Task The_store_outcome_is_surfaced_unchanged(ExperienceStoreOutcome stored, LifecycleTransitionOutcome expected) + { + var store = new RecordingStore + { + Result = new ExperienceLifecycleCommitResult(stored, 11, ExperienceStatus.Quarantined, [new StoreValidationError("Reason", "must not be empty or whitespace.")]), + }; + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(ExperienceStatus.Candidate, ExperienceStatus.Validated), CancellationToken.None); + + Assert.Equal(expected, result.Outcome); + Assert.Equal(11, result.Revision); + Assert.Equal(ExperienceStatus.Quarantined, result.CurrentStatus); + Assert.Equal("Reason", Assert.Single(result.Errors).Path); + } + + [Fact] + public async Task An_undefined_status_is_left_to_the_store_to_report_as_Invalid_not_refused_as_a_transition() + { + var store = new RecordingStore + { + Result = new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.Invalid, 0, null, [new StoreValidationError("CurrentStatus", "is not a defined value.")]), + }; + var service = new ExperienceLifecycleService(store); + + var result = await service.CommitAsync(Authorization, Request(ExperienceStatus.Candidate, (ExperienceStatus)999), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.Invalid, result.Outcome); + Assert.Equal("CurrentStatus", Assert.Single(result.Errors).Path); + Assert.Single(store.Commits); + } + + [Fact] + public async Task A_store_outcome_that_is_not_a_commit_outcome_is_never_reinterpreted() + { + var store = new RecordingStore { Result = new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.Created, 0, null, []) }; + var service = new ExperienceLifecycleService(store); + + await Assert.ThrowsAsync( + () => service.CommitAsync(Authorization, Request(ExperienceStatus.Candidate, ExperienceStatus.Validated), CancellationToken.None)); + } + + [Fact] + public async Task Infrastructure_failures_and_cancellation_propagate_from_the_port() + { + var failing = new RecordingStore { Throw = () => new ExperienceStoreException("storage failed") }; + await Assert.ThrowsAsync( + () => new ExperienceLifecycleService(failing).CommitAsync(Authorization, Request(ExperienceStatus.Candidate, ExperienceStatus.Revoked), CancellationToken.None)); + + var cancelling = new RecordingStore { Throw = () => new OperationCanceledException() }; + await Assert.ThrowsAnyAsync( + () => new ExperienceLifecycleService(cancelling).CommitAsync(Authorization, Request(ExperienceStatus.Candidate, ExperienceStatus.Revoked), CancellationToken.None)); + } + + [Fact] + public async Task A_null_prior_status_stamps_a_first_event_and_skips_the_transition_table() + { + var store = new RecordingStore(); + var service = new ExperienceLifecycleService(store); + + // Candidate -> Candidate is not in the table, but with no prior status there is no transition to + // look up: this is a record's first event, and the store skips its status match too. + var result = await service.CommitAsync(Authorization, Request(null, ExperienceStatus.Candidate), CancellationToken.None); + + Assert.Equal(LifecycleTransitionOutcome.Committed, result.Outcome); + Assert.Null(Assert.Single(store.Commits).Event.PriorStatus); + Assert.Null(result.Event!.PriorStatus); + } + + [Fact] + public async Task Null_arguments_throw_ArgumentNullException() + { + var service = new ExperienceLifecycleService(new RecordingStore()); + + Assert.Throws(() => new ExperienceLifecycleService(null!)); + await Assert.ThrowsAsync( + () => service.CommitAsync(null!, Request(ExperienceStatus.Candidate, ExperienceStatus.Revoked), CancellationToken.None)); + await Assert.ThrowsAsync(() => service.CommitAsync(Authorization, null!, CancellationToken.None)); + + // A null Scope is rejected here, not left to throw from inside the port. + var noScope = Request(ExperienceStatus.Candidate, ExperienceStatus.Revoked) with { Scope = null! }; + await Assert.ThrowsAsync(() => service.CommitAsync(Authorization, noScope, CancellationToken.None)); + } + + /// + /// Records what the service handed the port, and answers with a configurable outcome. Every other + /// port operation is out of this story's scope and fails loudly if the service ever calls it. + /// + private sealed class RecordingStore : IExperienceRecordStore + { + public List<(Scope Scope, LifecycleEvent Event)> Commits { get; } = []; + + public ExperienceLifecycleCommitResult? Result { get; init; } + + public Func? Throw { get; init; } + + public Task CommitLifecycleEventAsync( + AuthorizationContext authorization, + Scope scope, + LifecycleEvent lifecycleEvent, + CancellationToken cancellationToken) + { + Assert.NotNull(authorization); + Commits.Add((scope, lifecycleEvent)); + + if (Throw is not null) + { + throw Throw(); + } + + return Task.FromResult(Result + ?? new ExperienceLifecycleCommitResult(ExperienceStoreOutcome.Committed, lifecycleEvent.ExpectedRevision + 1, null, [])); + } + + public Task CreateAsync(AuthorizationContext authorization, ExperienceRecord record, CancellationToken cancellationToken) => + throw new InvalidOperationException("The lifecycle service must not create records."); + + public Task GetAsync(AuthorizationContext authorization, Scope scope, Guid experienceId, CancellationToken cancellationToken) => + throw new InvalidOperationException("The lifecycle service must not read records."); + + public Task QueryAsync(AuthorizationContext authorization, ExperienceRecordQuery query, CancellationToken cancellationToken) => + throw new InvalidOperationException("The lifecycle service must not query records."); + + public Task GetHistoryAsync(AuthorizationContext authorization, Scope scope, Guid experienceId, CancellationToken cancellationToken) => + throw new InvalidOperationException("The lifecycle service must not read history."); + } +} diff --git a/tests/AgentExperience.Storage.Postgres.Tests/OfflineStoreTests.cs b/tests/AgentExperience.Storage.Postgres.Tests/OfflineStoreTests.cs index 2661ada..5d8f2f4 100644 --- a/tests/AgentExperience.Storage.Postgres.Tests/OfflineStoreTests.cs +++ b/tests/AgentExperience.Storage.Postgres.Tests/OfflineStoreTests.cs @@ -313,11 +313,147 @@ public void Embedded_schema_script_is_available_and_creates_the_versioned_table( { var sql = PostgresExperienceRecordSchema.GetScript(PostgresExperienceRecordSchema.InitialScriptName); - Assert.Equal([PostgresExperienceRecordSchema.InitialScriptName], PostgresExperienceRecordSchema.ScriptNames); + Assert.Equal( + [PostgresExperienceRecordSchema.InitialScriptName, PostgresExperienceRecordSchema.LifecycleEventsScriptName], + PostgresExperienceRecordSchema.ScriptNames); Assert.Contains("CREATE SCHEMA IF NOT EXISTS agent_experience", sql, StringComparison.Ordinal); Assert.Contains("payload_version", sql, StringComparison.Ordinal); Assert.Contains("jsonb", sql, StringComparison.Ordinal); Assert.DoesNotContain("vector", sql, StringComparison.OrdinalIgnoreCase); Assert.Throws(() => PostgresExperienceRecordSchema.GetScript("9999_missing.sql")); } + + [Fact] + public void Lifecycle_script_is_embedded_separately_and_never_edits_the_initial_one() + { + var lifecycle = PostgresExperienceRecordSchema.GetScript(PostgresExperienceRecordSchema.LifecycleEventsScriptName); + + // Scripts are append-only: 0002 adds its own table and touches nothing 0001 created. + Assert.Contains("CREATE TABLE IF NOT EXISTS agent_experience.lifecycle_events", lifecycle, StringComparison.Ordinal); + Assert.Contains("applied_revision = expected_revision + 1", lifecycle, StringComparison.Ordinal); + // Unique, so two events can never claim one revision of a record and desynchronize the log. + Assert.Contains("CREATE UNIQUE INDEX IF NOT EXISTS ix_lifecycle_events_record_revision", lifecycle, StringComparison.Ordinal); + Assert.DoesNotContain("ALTER TABLE", lifecycle, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("DROP", lifecycle, StringComparison.OrdinalIgnoreCase); + + // 0002 is applied after 0001, which the migrator relies on for ordinal name ordering. + Assert.Equal( + PostgresExperienceRecordSchema.ScriptNames.Order(StringComparer.Ordinal), + PostgresExperienceRecordSchema.ScriptNames); + } + + [Fact] + public async Task Malformed_lifecycle_commit_returns_Invalid_with_every_field_path_and_no_database_call() + { + var tenant = NewTenant(); + var malformed = Event(Guid.Empty, (ExperienceStatus)999, (ExperienceStatus)998, -1, eventId: Guid.Empty, reason: " ", producer: "") + with { OccurredAt = default }; + + var result = await Store.CommitLifecycleEventAsync( + Authorize(tenant), + new Scope(tenant, "app-1", " "), + malformed, + CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Invalid, result.Outcome); + Assert.Equal(0, result.Revision); + Assert.Equal( + new[] { "CurrentStatus", "EventId", "ExperienceRecordId", "ExpectedRevision", "OccurredAt", "PriorStatus", "Producer", "Reason", "Scope.ProjectId" } + .Order(StringComparer.Ordinal), + result.Errors.Select(e => e.Path).Order(StringComparer.Ordinal)); + Assert.All(result.Errors, e => Assert.False(string.IsNullOrWhiteSpace(e.Message))); + } + + [Theory] + [InlineData(long.MaxValue)] // ExpectedRevision + 1 would wrap + [InlineData(long.MaxValue - 1)] // commits, but then no later commit could ever be expressed + public async Task A_revision_that_leaves_no_room_for_the_next_one_is_Invalid(long expectedRevision) + { + var tenant = NewTenant(); + + var result = await Store.CommitLifecycleEventAsync( + Authorize(tenant), + Scope(tenant), + Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Revoked, expectedRevision), + CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Invalid, result.Outcome); + Assert.Equal(["ExpectedRevision"], result.Errors.Select(e => e.Path)); + } + + [Fact] + public async Task An_unset_OccurredAt_is_Invalid_because_it_is_part_of_the_replay_identity() + { + var tenant = NewTenant(); + var unset = Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0) with { OccurredAt = default }; + + var result = await Store.CommitLifecycleEventAsync(Authorize(tenant), Scope(tenant), unset, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Invalid, result.Outcome); + Assert.Equal(["OccurredAt"], result.Errors.Select(e => e.Path)); + } + + [Fact] + public async Task Malformed_history_request_returns_Invalid() + { + var tenant = NewTenant(); + + var result = await Store.GetHistoryAsync(Authorize(tenant), new Scope(tenant, "", "project-1"), Guid.Empty, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Invalid, result.Outcome); + Assert.Equal(["ExperienceId", "Scope.ApplicationId"], result.Errors.Select(e => e.Path).Order(StringComparer.Ordinal)); + Assert.Empty(result.Events); + Assert.Equal(0, result.Revision); + } + + [Fact] + public async Task A_scope_beyond_the_authorization_is_Denied_before_any_connection_opens() + { + var scope = Scope("tenant-b"); + var auth = Authorize("tenant-a"); + + var commit = await Store.CommitLifecycleEventAsync( + auth, scope, Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Validated, 0), CancellationToken.None); + var history = await Store.GetHistoryAsync(auth, scope, Guid.NewGuid(), CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Denied, commit.Outcome); + Assert.Empty(commit.Errors); + Assert.Equal(ExperienceStoreOutcome.Denied, history.Outcome); + Assert.Empty(history.Events); + } + + [Fact] + public async Task An_unavailable_database_throws_for_commit_and_history_and_a_pre_cancelled_token_does_not() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var scope = Scope(tenant); + var lifecycleEvent = Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + + var commit = await Assert.ThrowsAsync( + () => Store.CommitLifecycleEventAsync(auth, scope, lifecycleEvent, CancellationToken.None)); + var history = await Assert.ThrowsAsync( + () => Store.GetHistoryAsync(auth, scope, lifecycleEvent.ExperienceRecordId, CancellationToken.None)); + Assert.All([commit, history], ex => Assert.IsAssignableFrom(ex.InnerException)); + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + var cancelled = await Assert.ThrowsAnyAsync( + () => Store.CommitLifecycleEventAsync(auth, scope, lifecycleEvent, cts.Token)); + Assert.IsNotType(cancelled); + } + + [Fact] + public async Task Null_lifecycle_arguments_throw_ArgumentNullException() + { + var tenant = NewTenant(); + var scope = Scope(tenant); + var lifecycleEvent = Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + + await Assert.ThrowsAsync(() => Store.CommitLifecycleEventAsync(null!, scope, lifecycleEvent, CancellationToken.None)); + await Assert.ThrowsAsync(() => Store.CommitLifecycleEventAsync(Authorize(tenant), null!, lifecycleEvent, CancellationToken.None)); + await Assert.ThrowsAsync(() => Store.CommitLifecycleEventAsync(Authorize(tenant), scope, null!, CancellationToken.None)); + await Assert.ThrowsAsync(() => Store.GetHistoryAsync(null!, scope, Guid.NewGuid(), CancellationToken.None)); + await Assert.ThrowsAsync(() => Store.GetHistoryAsync(Authorize(tenant), null!, Guid.NewGuid(), CancellationToken.None)); + } } diff --git a/tests/AgentExperience.Storage.Postgres.Tests/PostgresLifecycleCommitTests.cs b/tests/AgentExperience.Storage.Postgres.Tests/PostgresLifecycleCommitTests.cs new file mode 100644 index 0000000..2baf738 --- /dev/null +++ b/tests/AgentExperience.Storage.Postgres.Tests/PostgresLifecycleCommitTests.cs @@ -0,0 +1,647 @@ +using Npgsql; +using static AgentExperience.Storage.Postgres.Tests.TestRecords; + +namespace AgentExperience.Storage.Postgres.Tests; + +/// +/// Story 2.4's atomic lifecycle commit against a real PostgreSQL 16 container: one transaction per +/// commit, idempotency by event ID, optimistic concurrency by expected revision, and an append-only +/// history. Each test uses its own random tenant, so tests sharing the container never see each other's +/// rows. +/// +[Collection(PostgresCollection.Name)] +public sealed class PostgresLifecycleCommitTests +{ + private readonly PostgresFixture _fixture; + private readonly PostgresExperienceRecordStore _store; + + public PostgresLifecycleCommitTests(PostgresFixture fixture) + { + _fixture = fixture; + _store = new PostgresExperienceRecordStore(fixture.DataSource); + } + + [Fact] + public async Task A_valid_commit_stores_the_event_updates_the_projection_and_raises_the_revision() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, record.Revision); + var result = await _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Committed, result.Outcome); + Assert.Equal(record.Revision + 1, result.Revision); + Assert.Empty(result.Errors); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Validated, stored.Status); + Assert.Equal(record.Revision + 1, stored.Revision); + Assert.True(stored.UpdatedAt > record.UpdatedAt); + + // The adapter persists only the decision it was given: nothing else about the record moves. + Assert.Equal(record.ReuseConfidence, stored.ReuseConfidence); + Assert.Equal(record.SupportingValidations, stored.SupportingValidations); + Assert.Equal(record.Contradictions, stored.Contradictions); + Assert.Equal(record.CreatedAt, stored.CreatedAt); + + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Equal(ExperienceStoreOutcome.Found, history.Outcome); + Assert.Equal(record.Revision + 1, history.Revision); + var only = Assert.Single(history.Events); + Assert.Equal(lifecycleEvent.EventId, only.EventId); + Assert.Equal(ExperienceStatus.Candidate, only.PriorStatus); + Assert.Equal(ExperienceStatus.Validated, only.CurrentStatus); + Assert.Equal(lifecycleEvent.Reason, only.Reason); + Assert.Equal(lifecycleEvent.Producer, only.Producer); + Assert.Equal(TimeSpan.Zero, only.OccurredAt.Offset); + } + + [Fact] + public async Task Replaying_an_identical_event_returns_the_original_outcome_and_writes_nothing() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + + var first = await _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, CancellationToken.None); + var replay = await _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, CancellationToken.None); + var replayAgain = await _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Committed, first.Outcome); + Assert.Equal(first.Outcome, replay.Outcome); + Assert.Equal(first.Revision, replay.Revision); + Assert.Equal(first.Outcome, replayAgain.Outcome); + + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Single(history.Events); + Assert.Equal(1, history.Revision); + } + + [Theory] + [InlineData("reason")] + [InlineData("producer")] + [InlineData("prior")] + [InlineData("current")] + [InlineData("occurred")] + [InlineData("revision")] + [InlineData("record")] + [InlineData("scope")] + public async Task A_stored_event_id_with_any_differing_field_is_Conflict_and_writes_nothing(string difference) + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + var other = Minimal(Scope(tenant, project: "project-2")); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CreateAsync(auth, other, CancellationToken.None); + + var original = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + Assert.Equal( + ExperienceStoreOutcome.Committed, + (await _store.CommitLifecycleEventAsync(auth, record.Scope, original, CancellationToken.None)).Outcome); + + var scope = record.Scope; + var diverged = difference switch + { + "reason" => original with { Reason = original.Reason + "!" }, + "producer" => original with { Producer = "someone-else" }, + "prior" => original with { PriorStatus = null }, + "current" => original with { CurrentStatus = ExperienceStatus.Revoked }, + "occurred" => original with { OccurredAt = original.OccurredAt.AddSeconds(1) }, + "revision" => original with { ExpectedRevision = 1 }, + "record" => original with { ExperienceRecordId = other.ExperienceId }, + _ => original, + }; + + if (difference == "scope") + { + scope = other.Scope; + } + + var result = await _store.CommitLifecycleEventAsync(auth, scope, diverged, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Conflict, result.Outcome); + Assert.Equal(0, result.Revision); + Assert.Empty(result.Errors); + + // Neither the stored event nor either record moved. + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Equal(original, Assert.Single(history.Events) with { OccurredAt = original.OccurredAt }); + Assert.Equal(1, history.Revision); + var otherHistory = await _store.GetHistoryAsync(auth, other.Scope, other.ExperienceId, CancellationToken.None); + Assert.Empty(otherHistory.Events); + Assert.Equal(0, otherHistory.Revision); + } + + [Fact] + public async Task An_event_id_stored_in_a_foreign_tenant_conflicts_without_revealing_anything() + { + var tenant = NewTenant(); + var foreignTenant = NewTenant(); + var record = Minimal(Scope(tenant)); + var foreignRecord = Minimal(Scope(foreignTenant)); + await _store.CreateAsync(Authorize(tenant), record, CancellationToken.None); + await _store.CreateAsync(Authorize(foreignTenant), foreignRecord, CancellationToken.None); + + var eventId = Guid.NewGuid(); + await _store.CommitLifecycleEventAsync( + Authorize(tenant), record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0, eventId), CancellationToken.None); + + var result = await _store.CommitLifecycleEventAsync( + Authorize(foreignTenant), + foreignRecord.Scope, + Event(foreignRecord.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0, eventId), + CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Conflict, result.Outcome); + Assert.Empty(result.Errors); + var foreignHistory = await _store.GetHistoryAsync(Authorize(foreignTenant), foreignRecord.Scope, foreignRecord.ExperienceId, CancellationToken.None); + Assert.Empty(foreignHistory.Events); + Assert.Equal(0, foreignHistory.Revision); + } + + [Fact] + public async Task A_stale_expected_revision_writes_nothing_and_reports_the_current_revision() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0), CancellationToken.None); + + var behind = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + var ahead = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Validated, ExperienceStatus.Revoked, 5), CancellationToken.None); + + Assert.All([behind, ahead], result => + { + Assert.Equal(ExperienceStoreOutcome.StaleRevision, result.Outcome); + Assert.Equal(1, result.Revision); + Assert.Empty(result.Errors); + }); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Validated, stored.Status); + Assert.Equal(1, stored.Revision); + Assert.Single((await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Events); + } + + [Fact] + public async Task An_unknown_record_and_one_in_another_scope_are_both_NotFound() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant, project: "project-1", team: "team-1")); + await _store.CreateAsync(auth, record, CancellationToken.None); + + var missing = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(Guid.NewGuid(), ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + var foreignScope = await _store.CommitLifecycleEventAsync( + auth, Scope(tenant, project: "project-2", team: "team-1"), Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + var noTeam = await _store.CommitLifecycleEventAsync( + auth, Scope(tenant, project: "project-1"), Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + + Assert.All([missing, foreignScope, noTeam], result => + { + Assert.Equal(ExperienceStoreOutcome.NotFound, result.Outcome); + Assert.Equal(0, result.Revision); + Assert.Empty(result.Errors); + }); + + // A foreign-scope attempt must not leave an orphan event behind. + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Candidate, stored.Status); + Assert.Equal(0, stored.Revision); + Assert.Empty((await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Events); + Assert.Equal(0, await CountEventsAsync(record.ExperienceId)); + } + + [Fact] + public async Task History_of_a_record_in_another_scope_is_NotFound_like_a_missing_one() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant, team: "team-1")); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + + var otherTeam = await _store.GetHistoryAsync(auth, Scope(tenant, team: "team-2"), record.ExperienceId, CancellationToken.None); + var missing = await _store.GetHistoryAsync(auth, record.Scope, Guid.NewGuid(), CancellationToken.None); + + Assert.All([otherTeam, missing], result => + { + Assert.Equal(ExperienceStoreOutcome.NotFound, result.Outcome); + Assert.Equal(0, result.Revision); + Assert.Empty(result.Events); + Assert.Empty(result.Errors); + }); + } + + [Fact] + public async Task History_returns_every_step_oldest_first_with_the_record_s_current_revision() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + var validated = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + var quarantined = Event(record.ExperienceId, ExperienceStatus.Validated, ExperienceStatus.Quarantined, 1, reason: "sanitization gap suspected"); + var revoked = Event(record.ExperienceId, ExperienceStatus.Quarantined, ExperienceStatus.Revoked, 2, reason: "withdrawn by policy", producer: "governance"); + + foreach (var step in new[] { validated, quarantined, revoked }) + { + Assert.Equal( + ExperienceStoreOutcome.Committed, + (await _store.CommitLifecycleEventAsync(auth, record.Scope, step, CancellationToken.None)).Outcome); + } + + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Found, history.Outcome); + Assert.Equal(3, history.Revision); + Assert.Equal([validated.EventId, quarantined.EventId, revoked.EventId], history.Events.Select(e => e.EventId)); + Assert.Equal( + [ + (ExperienceStatus.Candidate, ExperienceStatus.Validated), + (ExperienceStatus.Validated, ExperienceStatus.Quarantined), + (ExperienceStatus.Quarantined, ExperienceStatus.Revoked), + ], + history.Events.Select(e => (e.PriorStatus, e.CurrentStatus))); + Assert.Equal([0L, 1L, 2L], history.Events.Select(e => e.ExpectedRevision)); + Assert.Equal("withdrawn by policy", history.Events[^1].Reason); + Assert.Equal("governance", history.Events[^1].Producer); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Revoked, stored.Status); + Assert.Equal(3, stored.Revision); + } + + [Fact] + public async Task History_of_a_record_with_no_events_is_Found_and_empty() + { + var tenant = NewTenant(); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(Authorize(tenant), record, CancellationToken.None); + + var history = await _store.GetHistoryAsync(Authorize(tenant), record.Scope, record.ExperienceId, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Found, history.Outcome); + Assert.Equal(0, history.Revision); + Assert.Empty(history.Events); + } + + [Fact] + public async Task A_first_lifecycle_event_may_carry_a_null_prior_status() + { + var tenant = NewTenant(); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(Authorize(tenant), record, CancellationToken.None); + + var result = await _store.CommitLifecycleEventAsync( + Authorize(tenant), record.Scope, Event(record.ExperienceId, null, ExperienceStatus.Quarantined, 0), CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Committed, result.Outcome); + var history = await _store.GetHistoryAsync(Authorize(tenant), record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Null(Assert.Single(history.Events).PriorStatus); + } + + [Fact] + public async Task Two_commits_from_the_same_revision_race_to_exactly_one_winner() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + var validate = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + var revoke = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0); + + var results = await Task.WhenAll( + _store.CommitLifecycleEventAsync(auth, record.Scope, validate, CancellationToken.None), + _store.CommitLifecycleEventAsync(auth, record.Scope, revoke, CancellationToken.None)); + + Assert.Equal(1, results.Count(r => r.Outcome == ExperienceStoreOutcome.Committed)); + Assert.Equal(1, results.Count(r => r.Outcome == ExperienceStoreOutcome.StaleRevision)); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(1, stored.Revision); + + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Equal(1, history.Revision); + var winner = Assert.Single(history.Events); + Assert.Equal(winner.CurrentStatus, stored.Status); + + // The loser's event never reached the log, so the log matches the projection exactly. + Assert.Equal(1, await CountEventsAsync(record.ExperienceId)); + } + + [Fact] + public async Task A_failure_after_the_event_insert_persists_neither_write() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + // The trigger below fires only for this task id, so no other test in the collection is affected. + var record = Minimal(Scope(tenant)) with { TaskId = "fail-mid-commit" }; + await _store.CreateAsync(auth, record, CancellationToken.None); + + await ExecuteAsync(""" + CREATE OR REPLACE FUNCTION agent_experience.fail_mid_commit() RETURNS trigger AS $body$ + BEGIN + RAISE EXCEPTION 'deliberate failure between the event insert and the projection update'; + END; + $body$ LANGUAGE plpgsql; + + CREATE TRIGGER fail_mid_commit + BEFORE UPDATE ON agent_experience.experience_records + FOR EACH ROW WHEN (NEW.task_id = 'fail-mid-commit') + EXECUTE FUNCTION agent_experience.fail_mid_commit(); + """); + + try + { + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + + var ex = await Assert.ThrowsAsync( + () => _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, CancellationToken.None)); + Assert.IsAssignableFrom(ex.InnerException); + + // Neither write survives: no event row, and the projection is untouched. + Assert.Equal(0, await CountEventsAsync(record.ExperienceId)); + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Candidate, stored.Status); + Assert.Equal(0, stored.Revision); + Assert.Equal(record.UpdatedAt, stored.UpdatedAt); + } + finally + { + await ExecuteAsync( + "DROP TRIGGER IF EXISTS fail_mid_commit ON agent_experience.experience_records; " + + "DROP FUNCTION IF EXISTS agent_experience.fail_mid_commit();"); + } + + // With the trigger gone the same event id commits normally, proving nothing was left half-written. + var retry = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0), CancellationToken.None); + Assert.Equal(ExperienceStoreOutcome.Committed, retry.Outcome); + Assert.Equal(1, retry.Revision); + } + + [Fact] + public async Task Cancelling_a_commit_mid_flight_surfaces_an_unwrapped_OperationCanceledException_and_writes_nothing() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + // Hold the record's row so the commit's projection update blocks after its event insert, giving + // the token a real in-flight command to cancel (rather than the pre-flight guard). + await using var blocker = await _fixture.DataSource.OpenConnectionAsync(); + var blocking = await blocker.BeginTransactionAsync(); + await using (var hold = new NpgsqlCommand( + "SELECT revision FROM agent_experience.experience_records WHERE experience_id = @id FOR UPDATE", blocker, blocking)) + { + hold.Parameters.Add(new NpgsqlParameter("id", record.ExperienceId)); + Assert.Equal(0L, await hold.ExecuteScalarAsync()); + } + + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(500)); + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + + var ex = await Assert.ThrowsAnyAsync( + () => _store.CommitLifecycleEventAsync(auth, record.Scope, lifecycleEvent, cts.Token)); + Assert.IsNotType(ex); + } + finally + { + await blocking.RollbackAsync(); + await blocking.DisposeAsync(); + } + + Assert.Equal(0, await CountEventsAsync(record.ExperienceId)); + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Candidate, stored.Status); + Assert.Equal(0, stored.Revision); + } + + [Fact] + public async Task Stored_event_rows_carry_the_scope_the_statuses_and_both_revisions() + { + var tenant = NewTenant(); + var scope = new Scope(tenant, "app-1", "project-1", "team-1", null, "user-1"); + var record = Minimal(scope); + await _store.CreateAsync(Authorize(tenant), record, CancellationToken.None); + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + await _store.CommitLifecycleEventAsync(Authorize(tenant), scope, lifecycleEvent, CancellationToken.None); + + await using var command = _fixture.DataSource.CreateCommand( + "SELECT tenant_id, team_id, agent_id, user_id, prior_status, current_status, expected_revision, applied_revision, " + + "occurred_at, recorded_at FROM agent_experience.lifecycle_events WHERE event_id = @id"); + command.Parameters.Add(new NpgsqlParameter("id", lifecycleEvent.EventId)); + await using var reader = await command.ExecuteReaderAsync(); + Assert.True(await reader.ReadAsync()); + + Assert.Equal(tenant, reader.GetString(0)); + Assert.Equal("team-1", reader.GetString(1)); + Assert.True(reader.IsDBNull(2)); + Assert.Equal("user-1", reader.GetString(3)); + Assert.Equal("Candidate", reader.GetString(4)); + Assert.Equal("Validated", reader.GetString(5)); + Assert.Equal(0L, reader.GetInt64(6)); + Assert.Equal(1L, reader.GetInt64(7)); + // Sub-microsecond ticks are truncated on write, exactly as the record's own columns are. + Assert.Equal(PayloadTime.AddTicks(-1), reader.GetFieldValue(8)); + Assert.True(reader.GetFieldValue(9) >= reader.GetFieldValue(8)); + } + + [Fact] + public async Task The_schema_rejects_an_event_that_bypasses_the_store() + { + // The projection and the log can only stay consistent if applied_revision follows expected_revision. + await using var command = _fixture.DataSource.CreateCommand( + "INSERT INTO agent_experience.lifecycle_events (event_id, experience_id, tenant_id, application_id, project_id, " + + "prior_status, current_status, reason, producer, occurred_at, recorded_at, expected_revision, applied_revision) " + + "VALUES (gen_random_uuid(), gen_random_uuid(), 'tenant', 'app', 'proj', NULL, 'Revoked', 'because', 'tests', now(), now(), 3, 7)"); + + var ex = await Assert.ThrowsAsync(() => command.ExecuteNonQueryAsync()); + Assert.Equal(PostgresErrorCodes.CheckViolation, ex.SqlState); + + await using var blank = _fixture.DataSource.CreateCommand( + "INSERT INTO agent_experience.lifecycle_events (event_id, experience_id, tenant_id, application_id, project_id, " + + "prior_status, current_status, reason, producer, occurred_at, recorded_at, expected_revision, applied_revision) " + + "VALUES (gen_random_uuid(), gen_random_uuid(), ' ', 'app', 'proj', NULL, 'Revoked', 'because', 'tests', now(), now(), 0, 1)"); + + var blankTenant = await Assert.ThrowsAsync(() => blank.ExecuteNonQueryAsync()); + Assert.Equal(PostgresErrorCodes.CheckViolation, blankTenant.SqlState); + } + + [Fact] + public async Task A_corrupt_stored_status_throws_ExperienceStoreException_on_history() + { + var tenant = NewTenant(); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(Authorize(tenant), record, CancellationToken.None); + var lifecycleEvent = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + await _store.CommitLifecycleEventAsync(Authorize(tenant), record.Scope, lifecycleEvent, CancellationToken.None); + + await using (var corrupt = _fixture.DataSource.CreateCommand( + "UPDATE agent_experience.lifecycle_events SET current_status = 'validated' WHERE event_id = @id")) + { + corrupt.Parameters.Add(new NpgsqlParameter("id", lifecycleEvent.EventId)); + Assert.Equal(1, await corrupt.ExecuteNonQueryAsync()); + } + + var ex = await Assert.ThrowsAsync( + () => _store.GetHistoryAsync(Authorize(tenant), record.Scope, record.ExperienceId, CancellationToken.None)); + + // The message must name the row that is actually corrupt, not the record. + Assert.Equal("Stored lifecycle event has an unrecognized status.", ex.Message); + } + + [Fact] + public async Task A_prior_status_the_record_is_not_in_writes_nothing_and_reports_the_stored_status() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + // Move the record to Quarantined, so its stored status no longer matches what a stale caller holds. + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Quarantined, 0), CancellationToken.None); + + // Revision 1 is correct, but the record is Quarantined, not Candidate. Without the prior-status + // guard this would commit a Candidate -> Validated transition Core forbids from Quarantined, and + // store a prior status the record never had. + var dishonest = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 1); + var result = await _store.CommitLifecycleEventAsync(auth, record.Scope, dishonest, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.StatusMismatch, result.Outcome); + Assert.Equal(1, result.Revision); + Assert.Equal(ExperienceStatus.Quarantined, result.CurrentStatus); + Assert.Empty(result.Errors); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Quarantined, stored.Status); + Assert.Equal(1, stored.Revision); + Assert.Equal(1, await CountEventsAsync(record.ExperienceId)); + } + + [Fact] + public async Task A_transition_Core_forbids_is_also_refused_by_the_store_when_asserted_dishonestly() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + + // Revoked -> Quarantined is outside Core's table. A caller that routes around Core by asserting a + // prior status the record is not in still cannot commit it. + var result = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Validated, ExperienceStatus.Quarantined, 1), CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.StatusMismatch, result.Outcome); + Assert.Equal(ExperienceStatus.Revoked, result.CurrentStatus); + + var stored = (await _store.GetAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None)).Record!; + Assert.Equal(ExperienceStatus.Revoked, stored.Status); + Assert.Equal(1, stored.Revision); + Assert.Equal(1, await CountEventsAsync(record.ExperienceId)); + } + + [Fact] + public async Task A_stale_revision_is_reported_even_when_the_prior_status_also_differs() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Revoked, 0), CancellationToken.None); + + // Both guards fail; the revision is the one the caller must fix first. + var result = await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Validated, ExperienceStatus.Quarantined, 0), CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.StaleRevision, result.Outcome); + Assert.Equal(1, result.Revision); + Assert.Null(result.CurrentStatus); + } + + [Fact] + public async Task Replaying_an_event_after_a_later_commit_reports_its_own_revision_not_the_current_one() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + + var first = Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0); + var second = Event(record.ExperienceId, ExperienceStatus.Validated, ExperienceStatus.Quarantined, 1); + var third = Event(record.ExperienceId, ExperienceStatus.Quarantined, ExperienceStatus.Revoked, 2); + foreach (var step in new[] { first, second, third }) + { + Assert.Equal( + ExperienceStoreOutcome.Committed, + (await _store.CommitLifecycleEventAsync(auth, record.Scope, step, CancellationToken.None)).Outcome); + } + + // The record is now at revision 3, so 1 can only come from the stored event's own applied revision. + var replay = await _store.CommitLifecycleEventAsync(auth, record.Scope, first, CancellationToken.None); + + Assert.Equal(ExperienceStoreOutcome.Committed, replay.Outcome); + Assert.Equal(1, replay.Revision); + Assert.Null(replay.CurrentStatus); + + var middle = await _store.CommitLifecycleEventAsync(auth, record.Scope, second, CancellationToken.None); + Assert.Equal(2, middle.Revision); + + // Nothing was written by either replay. + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + Assert.Equal(3, history.Revision); + Assert.Equal([first.EventId, second.EventId, third.EventId], history.Events.Select(e => e.EventId)); + } + + [Fact] + public async Task History_reports_a_revision_and_events_from_one_snapshot() + { + var tenant = NewTenant(); + var auth = Authorize(tenant); + var record = Minimal(Scope(tenant)); + await _store.CreateAsync(auth, record, CancellationToken.None); + await _store.CommitLifecycleEventAsync( + auth, record.Scope, Event(record.ExperienceId, ExperienceStatus.Candidate, ExperienceStatus.Validated, 0), CancellationToken.None); + + // Whatever else is happening, the revision must equal the highest applied revision in the events. + var history = await _store.GetHistoryAsync(auth, record.Scope, record.ExperienceId, CancellationToken.None); + + Assert.Equal(history.Events.Max(e => e.ExpectedRevision) + 1, history.Revision); + } + + private async Task CountEventsAsync(Guid experienceId) + { + await using var command = _fixture.DataSource.CreateCommand( + "SELECT count(*) FROM agent_experience.lifecycle_events WHERE experience_id = @id"); + command.Parameters.Add(new NpgsqlParameter("id", experienceId)); + return (long)(await command.ExecuteScalarAsync())!; + } + + private async Task ExecuteAsync(string sql) + { + await using var command = _fixture.DataSource.CreateCommand(sql); + await command.ExecuteNonQueryAsync(); + } +} diff --git a/tests/AgentExperience.Storage.Postgres.Tests/TestRecords.cs b/tests/AgentExperience.Storage.Postgres.Tests/TestRecords.cs index 27012bf..0aff75b 100644 --- a/tests/AgentExperience.Storage.Postgres.Tests/TestRecords.cs +++ b/tests/AgentExperience.Storage.Postgres.Tests/TestRecords.cs @@ -44,6 +44,24 @@ public static NpgsqlDataSource Unreachable() => CreatedAt: createdAt ?? ColumnTime, UpdatedAt: createdAt ?? ColumnTime); + /// A lifecycle event for , with a fresh event ID unless one is given. + public static LifecycleEvent Event( + Guid recordId, + ExperienceStatus? prior, + ExperienceStatus current, + long expectedRevision, + Guid? eventId = null, + string reason = "verified evidence", + string producer = "finalization") => new( + EventId: eventId ?? Guid.NewGuid(), + ExperienceRecordId: recordId, + PriorStatus: prior, + CurrentStatus: current, + Reason: reason, + Producer: producer, + OccurredAt: PayloadTime, + ExpectedRevision: expectedRevision); + /// A record with every optional part populated, including nested tool-call argument shapes. public static ExperienceRecord Full(Scope scope) {