nexus/backend-tests/OpenClawEventSubscriptionCoordinatorTests.cs

using System.Text.Json.Nodes;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Nexus.Api.Data;
using Nexus.Api.Repositories;
using Nexus.Api.Services;
using Xunit;

namespace Nexus.Api.Tests;

public sealed class OpenClawEventSubscriptionCoordinatorTests
{
    [Fact]
    public async Task SynchronizeOnce_SubscribesCatalogAndActiveSessions_OncePerConnection()
    {
        var services = new ServiceCollection();
        var databaseName = $"subscription-{Guid.NewGuid():N}";
        services.AddDbContext<NexusDbContext>(options =>
            options.UseInMemoryDatabase(databaseName));
        services.AddScoped<IOpenClawRunRepository, OpenClawRunRepository>();
        await using var provider = services.BuildServiceProvider();
        await using (var scope = provider.CreateAsyncScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
            db.OpenClawRuns.Add(new OpenClawRun
            {
                Title = "Active run",
                Prompt = "Do work",
                AgentId = "iris",
                SessionKey = "agent:iris:main",
                Status = OpenClawRunStates.Running,
                StartIdempotencyKey = "subscription-key",
                CorrelationId = "subscription-correlation",
                Actor = "owner"
            });
            await db.SaveChangesAsync();
        }

        var connector = new SubscriptionConnector();
        var coordinator = new OpenClawEventSubscriptionCoordinator(
            connector,
            provider.GetRequiredService<IServiceScopeFactory>(),
            NullLogger<OpenClawEventSubscriptionCoordinator>.Instance);

        await coordinator.SynchronizeOnceAsync();
        await coordinator.SynchronizeOnceAsync();

        Assert.Equal(2, connector.Calls.Count);
        Assert.Equal("sessions.subscribe", connector.Calls[0].Method);
        Assert.Equal("sessions.messages.subscribe", connector.Calls[1].Method);
        Assert.Equal("agent:iris:main", connector.Calls[1].Parameters?["key"]?.GetValue<string>());
        Assert.Equal("iris", connector.Calls[1].Parameters?["agentId"]?.GetValue<string>());
        Assert.True(connector.Calls[1].Parameters?["includeApprovals"]?.GetValue<bool>());

        await SetRunStatusAsync(provider, OpenClawRunStates.Completed);
        await coordinator.SynchronizeOnceAsync();
        Assert.Equal(3, connector.Calls.Count);
        Assert.Equal("sessions.messages.unsubscribe", connector.Calls[2].Method);

        await SetRunStatusAsync(provider, OpenClawRunStates.Running);
        connector.LastConnectedAtValue = connector.LastConnectedAtValue.AddMinutes(1);
        await coordinator.SynchronizeOnceAsync();

        Assert.Equal(5, connector.Calls.Count);
        Assert.Equal("sessions.subscribe", connector.Calls[3].Method);
        Assert.Equal("sessions.messages.subscribe", connector.Calls[4].Method);
    }

    private static async Task SetRunStatusAsync(
        ServiceProvider provider,
        string status)
    {
        await using var scope = provider.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
        var run = await db.OpenClawRuns.SingleAsync();
        run.Status = status;
        await db.SaveChangesAsync();
    }

    private sealed class SubscriptionConnector : IGatewayConnector
    {
        public List<SubscriptionCall> Calls { get; } = [];
        public DateTimeOffset LastConnectedAtValue { get; set; } = DateTimeOffset.UtcNow;
        public GatewayConnectionState ConnectionState => GatewayConnectionState.Connected;
        public string? GatewayVersion => "2026.7.0";
        public string? RequiredVersion => "2026.7.0";
        public DateTimeOffset? LastConnectedAt => LastConnectedAtValue;
        public int ReconnectAttempts => 0;
        public string? StatusMessage => "Connected";
        public string? DeviceId => "nexus-test";
        public bool DeviceTokenConfigured => true;
        public bool PairingRequired => false;
        public string? PairingRequestId => null;
        public int? ProtocolVersion => 4;
        public IReadOnlySet<string> AdvertisedMethods { get; } = new HashSet<string>(
            [
                "sessions.subscribe",
                "sessions.messages.subscribe",
                "sessions.messages.unsubscribe"
            ],
            StringComparer.Ordinal);
        public IReadOnlySet<string> AdvertisedEvents { get; } = new HashSet<string>();
        public IReadOnlySet<string> GrantedScopes { get; } = new HashSet<string>(
            ["operator.read", "operator.approvals"],
            StringComparer.Ordinal);
        public DateTimeOffset? LastEventAt => null;

        public bool Supports(string method) => AdvertisedMethods.Contains(method);

        public Task<JsonNode?> InvokeAsync(
            string method,
            object? parameters = null,
            TimeSpan? timeout = null,
            CancellationToken cancellationToken = default,
            OpenClawInvocationContext? invocationContext = null)
        {
            Calls.Add(new SubscriptionCall(
                method,
                parameters as JsonNode,
                invocationContext));
            return Task.FromResult<JsonNode?>(new JsonObject { ["ok"] = true });
        }

        public IReadOnlyList<GatewayEventEnvelope> GetRecentEvents(int limit = 100) => [];
    }

    private sealed record SubscriptionCall(
        string Method,
        JsonNode? Parameters,
        OpenClawInvocationContext? Invocation);
}