nexus/backend-tests/PostgresOpenClawOperationAuditStoreTests.cs

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Nexus.Api.Data;
using Nexus.Api.Services;
using Xunit;

namespace Nexus.Api.Tests;

public sealed class PostgresOpenClawOperationAuditStoreTests
{
    [Fact]
    public async Task Completed_claim_replays_from_database_without_writing_jsonl()
    {
        await using var fixture = Fixture.Create();
        var context = Context("database-replay");
        var operation = Operation("cron.run", "job-1", "intent-a");

        var started = await fixture.Store.ClaimAsync(context, operation);
        await fixture.Store.CompleteAsync(
            context,
            operation,
            true,
            "enqueued",
            "This message is deliberately not persisted.");
        var replay = await fixture.Store.ClaimAsync(context, operation);

        Assert.Equal(OpenClawOperationClaimDisposition.Started, started.Disposition);
        Assert.Equal(OpenClawOperationClaimDisposition.Replayed, replay.Disposition);
        Assert.True(replay.PreviousOk);
        Assert.Equal("enqueued", replay.PreviousState);
        Assert.False(File.Exists(fixture.AuditPath));

        await using var scope = fixture.Provider.CreateAsyncScope();
        var claim = await scope.ServiceProvider
            .GetRequiredService<NexusDbContext>()
            .OperationClaims
            .SingleAsync();
        Assert.Equal("openclaw.mutation", claim.Operation);
        Assert.Equal("completed", claim.State);
        Assert.Equal("enqueued", claim.ResultCode);
        Assert.DoesNotContain("database-replay", claim.IdempotencyKeyHash);
        Assert.DoesNotContain("This message", claim.ResultCode);
    }

    [Fact]
    public async Task Same_key_with_different_intent_conflicts()
    {
        await using var fixture = Fixture.Create();
        var context = Context("conflict-key");

        await fixture.Store.ClaimAsync(
            context,
            Operation("cron.run", "job-1", "intent-a"));
        var conflict = await fixture.Store.ClaimAsync(
            context,
            Operation("cron.run", "job-2", "intent-b"));

        Assert.Equal(
            OpenClawOperationClaimDisposition.Conflict,
            conflict.Disposition);
    }

    [Fact]
    public async Task Expired_incomplete_claim_remains_in_doubt()
    {
        await using var fixture = Fixture.Create();
        var context = Context("in-doubt-key");
        var operation = Operation("agents.files.set", "agent/SOUL.md", "intent");
        await fixture.Store.ClaimAsync(context, operation);
        await fixture.ExpireClaimAsync();

        var replay = await fixture.Store.ClaimAsync(context, operation);

        Assert.Equal(
            OpenClawOperationClaimDisposition.InDoubt,
            replay.Disposition);
    }

    [Fact]
    public async Task Expired_terminal_claim_can_be_reclaimed()
    {
        await using var fixture = Fixture.Create();
        var context = Context("expired-terminal-key");
        var operation = Operation("config.patch", "primary", "intent");
        await fixture.Store.ClaimAsync(context, operation);
        await fixture.Store.CompleteAsync(
            context,
            operation,
            false,
            "conflict",
            "Not persisted.");
        await fixture.ExpireClaimAsync();

        var reclaimed = await fixture.Store.ClaimAsync(context, operation);

        Assert.Equal(
            OpenClawOperationClaimDisposition.Started,
            reclaimed.Disposition);
        await using var scope = fixture.Provider.CreateAsyncScope();
        Assert.Single(await scope.ServiceProvider
            .GetRequiredService<NexusDbContext>()
            .OperationClaims
            .ToArrayAsync());
    }

    [Fact]
    public async Task Database_failure_is_not_treated_as_a_started_claim()
    {
        var fixture = Fixture.Create();
        var store = fixture.Store;
        await fixture.DisposeAsync();

        await Assert.ThrowsAnyAsync<ObjectDisposedException>(() =>
            store.ClaimAsync(
                Context("database-down"),
                Operation("cron.run", "job-1", "intent")));
    }

    private static OpenClawInvocationContext Context(string key)
        => OpenClawInvocationContext.Create(
            actor: "owner",
            idempotencyKey: key,
            correlationId: $"correlation-{key}");

    private static OpenClawOperationDescriptor Operation(
        string method,
        string targetId,
        string intent)
        => new(
            method,
            "test-resource",
            targetId,
            OpenClawInvocationContextFactory.Hash(intent));

    private sealed class Fixture(
        ServiceProvider provider,
        PostgresOpenClawOperationAuditStore store,
        string auditPath) : IAsyncDisposable
    {
        public ServiceProvider Provider { get; } = provider;
        public PostgresOpenClawOperationAuditStore Store { get; } = store;
        public string AuditPath { get; } = auditPath;

        public static Fixture Create()
        {
            var services = new ServiceCollection();
            var databaseName = $"operation-claims-{Guid.NewGuid():N}";
            services.AddDbContext<NexusDbContext>(options =>
                options.UseInMemoryDatabase(databaseName));
            var provider = services.BuildServiceProvider();
            var auditPath = Path.Combine(
                Path.GetTempPath(),
                $"nexus-operation-archive-{Guid.NewGuid():N}.jsonl");
            var store = new PostgresOpenClawOperationAuditStore(
                provider.GetRequiredService<IServiceScopeFactory>(),
                Options.Create(new GatewayConnectorOptions
                {
                    OperationAuditPath = auditPath
                }));
            return new Fixture(provider, store, auditPath);
        }

        public async Task ExpireClaimAsync()
        {
            await using var scope = Provider.CreateAsyncScope();
            var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
            var claim = await db.OperationClaims.SingleAsync();
            claim.ExpiresAt = DateTimeOffset.UtcNow.AddMinutes(-1);
            await db.SaveChangesAsync();
        }

        public ValueTask DisposeAsync() => Provider.DisposeAsync();
    }
}