AzuTear
355 lines
13 KiB
C#
355 lines
13 KiB
C#
using System.Text.Json.Nodes;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using Npgsql;
|
|
using Nexus.Api.Data;
|
|
using Nexus.Api.Models;
|
|
using Nexus.Api.Services;
|
|
using Testcontainers.PostgreSql;
|
|
using Xunit;
|
|
|
|
namespace Nexus.Api.Tests;
|
|
|
|
[Collection(DockerIntegrationTestEnvironment.CollectionName)]
|
|
public sealed class PostgreSqlAgentProvisioningIntegrationTests
|
|
{
|
|
private const string Category = "DockerIntegration";
|
|
|
|
[PostgreSqlIntegrationFact]
|
|
[Trait("Category", Category)]
|
|
public async Task Migrations_create_postgres_17_schema_and_required_indexes()
|
|
{
|
|
await using var postgres = BuildPostgreSql();
|
|
await postgres.StartAsync();
|
|
|
|
await using var db = CreateDatabase(postgres.GetConnectionString());
|
|
await db.Database.MigrateAsync();
|
|
|
|
Assert.Empty(await db.Database.GetPendingMigrationsAsync());
|
|
Assert.Contains(
|
|
"20260730224500_AddAgentProvisioningAndBoardIndexes",
|
|
await db.Database.GetAppliedMigrationsAsync());
|
|
|
|
await using var connection = new NpgsqlConnection(
|
|
postgres.GetConnectionString());
|
|
await connection.OpenAsync();
|
|
|
|
await using var versionCommand = new NpgsqlCommand(
|
|
"SELECT current_setting('server_version_num')::integer",
|
|
connection);
|
|
var version = Convert.ToInt32(await versionCommand.ExecuteScalarAsync());
|
|
Assert.InRange(version, 170000, 179999);
|
|
|
|
await using var tableCommand = new NpgsqlCommand(
|
|
"""
|
|
SELECT count(*)
|
|
FROM pg_class
|
|
WHERE relnamespace = 'public'::regnamespace
|
|
AND relkind = 'r'
|
|
AND relname IN (
|
|
'AgentProposals',
|
|
'AgentProvisionRequests',
|
|
'OperationClaims',
|
|
'OutboxEvents')
|
|
""",
|
|
connection);
|
|
Assert.Equal(4L, Convert.ToInt64(await tableCommand.ExecuteScalarAsync()));
|
|
|
|
await using var indexCommand = new NpgsqlCommand(
|
|
"""
|
|
SELECT count(*)
|
|
FROM pg_indexes
|
|
WHERE schemaname = 'public'
|
|
AND indexname IN (
|
|
'IX_AgentProvisionRequests_Status_CreatedAt',
|
|
'IX_OperationClaims_Operation_IdempotencyKeyHash',
|
|
'IX_OutboxEvents_PublishedAt_Sequence',
|
|
'IX_Tasks_State_UpdatedAt_Id_Board',
|
|
'IX_Tasks_Done_UpdatedAt_Id',
|
|
'IX_Tasks_ParentTaskId_State',
|
|
'IX_Activity_TaskId_CreatedAt_Id')
|
|
""",
|
|
connection);
|
|
Assert.Equal(7L, Convert.ToInt64(await indexCommand.ExecuteScalarAsync()));
|
|
}
|
|
|
|
[PostgreSqlIntegrationFact]
|
|
[Trait("Category", Category)]
|
|
public async Task Concurrent_proposal_creates_share_one_claim_and_one_outbox_event()
|
|
{
|
|
await using var postgres = BuildPostgreSql();
|
|
await postgres.StartAsync();
|
|
var connectionString = postgres.GetConnectionString();
|
|
var gateway = new ConcurrentApprovalGatewayConnector(
|
|
synchronizeAgentLists: false);
|
|
await MigrateAndSeedManagementAsync(connectionString, gateway);
|
|
|
|
const string idempotencyKey = "postgres-concurrent-proposal";
|
|
var operations = Enumerable.Range(0, 8)
|
|
.Select(_ => CreateProposalAsync(
|
|
connectionString,
|
|
gateway,
|
|
idempotencyKey))
|
|
.ToArray();
|
|
var results = await Task.WhenAll(operations);
|
|
|
|
Assert.All(results, result => Assert.True(result.Ok));
|
|
Assert.Single(results.Select(result => result.Proposal!.Id).Distinct());
|
|
Assert.Contains(
|
|
results,
|
|
result => result.State == AgentProposalStates.AwaitingApproval);
|
|
Assert.Contains(results, result => result.State == "idempotent_replay");
|
|
|
|
await using var verification = CreateDatabase(connectionString);
|
|
var proposal = await verification.AgentProposals.SingleAsync();
|
|
var claim = await verification.OperationClaims.SingleAsync();
|
|
var outbox = await verification.OutboxEvents.SingleAsync();
|
|
|
|
Assert.Equal(proposal.Id, claim.ResourceId);
|
|
Assert.Equal("agent-proposal.create", claim.Operation);
|
|
Assert.Equal("completed", claim.State);
|
|
Assert.Equal(proposal.Id.ToString("N"), outbox.AggregateId);
|
|
Assert.Equal("agent.proposal.created", outbox.Type);
|
|
Assert.DoesNotContain(
|
|
idempotencyKey,
|
|
claim.IdempotencyKeyHash,
|
|
StringComparison.Ordinal);
|
|
|
|
await using var connection = new NpgsqlConnection(connectionString);
|
|
await connection.OpenAsync();
|
|
await using var jsonTypeCommand = new NpgsqlCommand(
|
|
"""SELECT pg_typeof("PayloadJson")::text FROM "OutboxEvents" LIMIT 1""",
|
|
connection);
|
|
Assert.Equal("jsonb", await jsonTypeCommand.ExecuteScalarAsync());
|
|
}
|
|
|
|
[PostgreSqlIntegrationFact]
|
|
[Trait("Category", Category)]
|
|
public async Task Concurrent_owner_approvals_queue_exactly_one_provision_request()
|
|
{
|
|
await using var postgres = BuildPostgreSql();
|
|
await postgres.StartAsync();
|
|
var connectionString = postgres.GetConnectionString();
|
|
var gateway = new ConcurrentApprovalGatewayConnector(
|
|
synchronizeAgentLists: true);
|
|
await MigrateAndSeedManagementAsync(connectionString, gateway);
|
|
|
|
AgentProposalOperationDto created;
|
|
await using (var creationDb = CreateDatabase(connectionString))
|
|
{
|
|
created = await CreateService(creationDb, gateway).CreateAsync(
|
|
Proposal(),
|
|
"manual",
|
|
Invocation("postgres-approval-proposal"));
|
|
}
|
|
|
|
var revision = created.Proposal!.Revision;
|
|
var approvals = await Task.WhenAll(
|
|
ApproveAsync(
|
|
connectionString,
|
|
gateway,
|
|
created.Proposal.Id,
|
|
revision,
|
|
"postgres-approval-a"),
|
|
ApproveAsync(
|
|
connectionString,
|
|
gateway,
|
|
created.Proposal.Id,
|
|
revision,
|
|
"postgres-approval-b"));
|
|
|
|
var succeeded = Assert.Single(approvals, result => result.Ok);
|
|
var conflicted = Assert.Single(approvals, result => !result.Ok);
|
|
Assert.Equal(AgentProposalStates.Provisioning, succeeded.State);
|
|
Assert.Equal("concurrency_conflict", conflicted.State);
|
|
|
|
await using var verification = CreateDatabase(connectionString);
|
|
var proposal = await verification.AgentProposals.SingleAsync();
|
|
var request = await verification.AgentProvisionRequests.SingleAsync();
|
|
var claims = await verification.OperationClaims
|
|
.OrderBy(item => item.CreatedAt)
|
|
.ToArrayAsync();
|
|
var events = await verification.OutboxEvents
|
|
.OrderBy(item => item.Sequence)
|
|
.ToArrayAsync();
|
|
|
|
Assert.Equal(AgentProposalStates.Provisioning, proposal.Status);
|
|
Assert.Equal(AgentProvisionRequestStates.Queued, request.Status);
|
|
Assert.Equal(1, request.Attempt);
|
|
Assert.Equal(3, claims.Length);
|
|
Assert.Single(
|
|
claims,
|
|
item => item.Operation == "agent-proposal.approve"
|
|
&& item.State == "completed");
|
|
Assert.Single(
|
|
claims,
|
|
item => item.Operation == "agent-proposal.approve"
|
|
&& item.ResultCode == "concurrency_conflict");
|
|
Assert.Single(
|
|
events,
|
|
item => item.Type == "agent.provision.requested");
|
|
}
|
|
|
|
private static PostgreSqlContainer BuildPostgreSql()
|
|
=> new PostgreSqlBuilder("postgres:17-alpine")
|
|
.WithDatabase("nexus_integration")
|
|
.WithUsername("nexus_test")
|
|
.WithPassword("nexus_test_password")
|
|
.Build();
|
|
|
|
private static NexusDbContext CreateDatabase(string connectionString)
|
|
=> new(new DbContextOptionsBuilder<NexusDbContext>()
|
|
.UseNpgsql(connectionString)
|
|
.EnableDetailedErrors()
|
|
.Options);
|
|
|
|
private static async Task MigrateAndSeedManagementAsync(
|
|
string connectionString,
|
|
IGatewayConnector gateway)
|
|
{
|
|
await using var db = CreateDatabase(connectionString);
|
|
await db.Database.MigrateAsync();
|
|
db.OpenClawConnectionProfiles.Add(new OpenClawConnectionProfile
|
|
{
|
|
Endpoint = "ws://127.0.0.1:18789/",
|
|
DiscoverySource = "testcontainers",
|
|
RequiredVersion = gateway.RequiredVersion,
|
|
AdoptionState = OpenClawAdoptionStates.Adopted,
|
|
ManagementEnabled = true,
|
|
CapabilityHash = AgentProposalService.BuildCapabilityHash(gateway),
|
|
Revision = 1
|
|
});
|
|
await db.SaveChangesAsync();
|
|
}
|
|
|
|
private static async Task<AgentProposalOperationDto> CreateProposalAsync(
|
|
string connectionString,
|
|
IGatewayConnector gateway,
|
|
string idempotencyKey)
|
|
{
|
|
await using var db = CreateDatabase(connectionString);
|
|
return await CreateService(db, gateway).CreateAsync(
|
|
Proposal(),
|
|
"manual",
|
|
Invocation(idempotencyKey));
|
|
}
|
|
|
|
private static async Task<AgentProposalOperationDto> ApproveAsync(
|
|
string connectionString,
|
|
IGatewayConnector gateway,
|
|
Guid proposalId,
|
|
int expectedRevision,
|
|
string idempotencyKey)
|
|
{
|
|
await using var db = CreateDatabase(connectionString);
|
|
return await CreateService(db, gateway).ApproveAsync(
|
|
proposalId,
|
|
new AgentProposalActionRequest(expectedRevision),
|
|
Invocation(idempotencyKey));
|
|
}
|
|
|
|
internal static AgentProposalService CreateService(
|
|
NexusDbContext db,
|
|
IGatewayConnector gateway,
|
|
IOpenClawAgentConfigurationService? agentFiles = null)
|
|
{
|
|
var management = new OpenClawManagementState();
|
|
management.SetEnabled(true);
|
|
var configuration = new ConfigurationBuilder()
|
|
.AddInMemoryCollection(new Dictionary<string, string?>
|
|
{
|
|
["OpenClawSetup:ExternalClientIdentitySupported"] = "true"
|
|
})
|
|
.Build();
|
|
return new AgentProposalService(
|
|
db,
|
|
gateway,
|
|
agentFiles ?? new ProposalAgentConfigurationService(),
|
|
new StubOpenClawWriteGate(),
|
|
Options.Create(new AgentProvisioningOptions()),
|
|
new AgentProvisioningSignal(),
|
|
NullLogger<AgentProposalService>.Instance);
|
|
}
|
|
|
|
internal static CreateAgentProposalRequest Proposal()
|
|
=> new(
|
|
"Release Analyst",
|
|
Role: "Release quality",
|
|
Description: "Verify releases and report evidence.",
|
|
Model: "openai/gpt-5.5",
|
|
ClientRequestId: "postgres-proposal");
|
|
|
|
internal static OpenClawInvocationMetadata Invocation(string key)
|
|
=> new(key, $"correlation-{key}", "bao", null);
|
|
}
|
|
|
|
internal sealed class ConcurrentApprovalGatewayConnector(
|
|
bool synchronizeAgentLists) : IGatewayConnector
|
|
{
|
|
private readonly TaskCompletionSource<bool> bothInventoryReads =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private int inventoryReads;
|
|
|
|
public GatewayConnectionState ConnectionState => GatewayConnectionState.Connected;
|
|
public string? GatewayVersion => "2026.7.1";
|
|
public string? RequiredVersion => "2026.7.1";
|
|
public DateTimeOffset? LastConnectedAt => DateTimeOffset.UtcNow;
|
|
public int ReconnectAttempts => 0;
|
|
public string? StatusMessage => "ready";
|
|
public string? DeviceId => "testcontainers-device";
|
|
public bool DeviceTokenConfigured => true;
|
|
public bool PairingRequired => false;
|
|
public string? PairingRequestId => null;
|
|
public int? ProtocolVersion => 4;
|
|
public IReadOnlySet<string> AdvertisedMethods { get; } =
|
|
new HashSet<string>(
|
|
[
|
|
"agents.list",
|
|
"agents.create",
|
|
"agents.files.get",
|
|
"agents.files.set",
|
|
"config.get"
|
|
],
|
|
StringComparer.Ordinal);
|
|
public IReadOnlySet<string> AdvertisedEvents { get; } =
|
|
new HashSet<string>(StringComparer.Ordinal);
|
|
public IReadOnlySet<string> GrantedScopes { get; } =
|
|
new HashSet<string>(
|
|
["operator.read", "operator.admin"],
|
|
StringComparer.Ordinal);
|
|
public DateTimeOffset? LastEventAt => null;
|
|
|
|
public bool Supports(string method) => AdvertisedMethods.Contains(method);
|
|
|
|
public async Task<JsonNode?> InvokeAsync(
|
|
string method,
|
|
object? parameters = null,
|
|
TimeSpan? timeout = null,
|
|
CancellationToken cancellationToken = default,
|
|
OpenClawInvocationContext? invocationContext = null)
|
|
{
|
|
if (method != "agents.list")
|
|
{
|
|
throw new OpenClawGatewayRpcException(
|
|
"METHOD_NOT_FOUND",
|
|
$"Unexpected integration-test method {method}.");
|
|
}
|
|
|
|
if (synchronizeAgentLists)
|
|
{
|
|
if (Interlocked.Increment(ref inventoryReads) == 2)
|
|
bothInventoryReads.TrySetResult(true);
|
|
await bothInventoryReads.Task.WaitAsync(
|
|
TimeSpan.FromSeconds(15),
|
|
cancellationToken);
|
|
}
|
|
|
|
return new JsonObject { ["agents"] = new JsonArray() };
|
|
}
|
|
|
|
public IReadOnlyList<GatewayEventEnvelope> GetRecentEvents(int limit = 100)
|
|
=> [];
|
|
}
|