nexus/backend-tests/ToxiproxyAgentProvisioningIntegrationTests.cs

using System.Net.Http.Json;
using System.Text.Json;
using System.Text.Json.Nodes;
using DotNet.Testcontainers.Builders;
using Microsoft.EntityFrameworkCore;
using Nexus.Api.Data;
using Nexus.Api.Models;
using Nexus.Api.Services;
using Testcontainers.Toxiproxy;
using Xunit;

namespace Nexus.Api.Tests;

[Collection(DockerIntegrationTestEnvironment.CollectionName)]
public sealed class ToxiproxyAgentProvisioningIntegrationTests
{
    private const ushort ToxiproxyAdminPort = 8474;
    private const ushort OpenClawProxyPort = 8666;

    [ToxiproxyIntegrationFact]
    [Trait("Category", "DockerIntegration")]
    [Trait("Category", "ToxiproxyIntegration")]
    public async Task Timeout_after_create_dispatch_becomes_in_doubt_without_duplicate_create()
    {
        await using var network = new NetworkBuilder().Build();
        await using var openClawStub = new ContainerBuilder(
                "python:3.13.7-alpine3.22")
            .WithNetwork(network)
            .WithNetworkAliases("openclaw-stub")
            .WithExposedPort(8080)
            .WithEntrypoint("python", "-u", "-c")
            .WithCommand(OpenClawStubScript)
            .WithWaitStrategy(
                Wait.ForUnixContainer()
                    .UntilInternalTcpPortIsAvailable(8080))
            .Build();
        await using var toxiproxy = new ToxiproxyBuilder(
                "ghcr.io/shopify/toxiproxy:2.12.0")
            .WithNetwork(network)
            .Build();

        await network.CreateAsync();
        await Task.WhenAll(
            openClawStub.StartAsync(),
            toxiproxy.StartAsync());

        using var toxiproxyAdmin = new HttpClient
        {
            BaseAddress = new Uri(
                $"http://{toxiproxy.Hostname}:"
                + toxiproxy.GetMappedPublicPort(ToxiproxyAdminPort))
        };
        using (var proxyResponse = await toxiproxyAdmin.PostAsJsonAsync(
                   "/proxies",
                   new
                   {
                       name = "openclaw",
                       listen = $"0.0.0.0:{OpenClawProxyPort}",
                       upstream = "openclaw-stub:8080",
                       enabled = true
                   }))
        {
            proxyResponse.EnsureSuccessStatusCode();
        }

        var gatewayEndpoint = new Uri(
            $"http://{toxiproxy.Hostname}:"
            + toxiproxy.GetMappedPublicPort(OpenClawProxyPort));
        using var gateway = new ToxiproxyHttpGatewayConnector(gatewayEndpoint);
        await using var db = new NexusDbContext(
            new DbContextOptionsBuilder<NexusDbContext>()
                .UseInMemoryDatabase($"toxiproxy-agent-{Guid.NewGuid():N}")
                .Options);
        await db.Database.EnsureCreatedAsync();
        var management = new OpenClawConnectionProfile
        {
            Endpoint = gatewayEndpoint.ToString(),
            DiscoverySource = "testcontainers-toxiproxy",
            RequiredVersion = gateway.RequiredVersion,
            AdoptionState = OpenClawAdoptionStates.Adopted,
            ManagementEnabled = true,
            CapabilityHash = AgentProposalService.BuildCapabilityHash(gateway),
            Revision = 1
        };
        db.OpenClawConnectionProfiles.Add(management);
        await db.SaveChangesAsync();

        var service = PostgreSqlAgentProvisioningIntegrationTests.CreateService(
            db,
            gateway);
        var created = await service.CreateAsync(
            PostgreSqlAgentProvisioningIntegrationTests.Proposal(),
            "manual",
            PostgreSqlAgentProvisioningIntegrationTests.Invocation(
                "toxiproxy-proposal"));
        var approved = await service.ApproveAsync(
            created.Proposal!.Id,
            new(created.Proposal.Revision),
            PostgreSqlAgentProvisioningIntegrationTests.Invocation(
                "toxiproxy-approval"));
        Assert.True(approved.Ok);

        gateway.BeforeCreateAsync = async cancellationToken =>
        {
            using var toxicResponse = await toxiproxyAdmin.PostAsJsonAsync(
                "/proxies/openclaw/toxics",
                new
                {
                    name = "delay-create-response",
                    type = "latency",
                    stream = "downstream",
                    toxicity = 1.0,
                    attributes = new { latency = 1500, jitter = 0 }
                },
                cancellationToken);
            toxicResponse.EnsureSuccessStatusCode();
        };

        Assert.True(await service.ProcessNextAsync());
        Assert.False(await service.ProcessNextAsync());

        var uncertain = await service.GetByIdAsync(created.Proposal.Id);
        Assert.NotNull(uncertain);
        Assert.Equal(AgentProposalStates.InDoubt, uncertain!.Status);
        Assert.Equal(1, gateway.CreateCalls);

        using (var removeToxic = await toxiproxyAdmin.DeleteAsync(
                   "/proxies/openclaw/toxics/delay-create-response"))
        {
            removeToxic.EnsureSuccessStatusCode();
        }

        var retry = await service.RetryAsync(
            created.Proposal.Id,
            new(uncertain.Revision),
            PostgreSqlAgentProvisioningIntegrationTests.Invocation(
                "toxiproxy-reconcile"));
        Assert.True(retry.Ok);
        Assert.True(await service.ProcessNextAsync());

        var reconciled = await service.GetByIdAsync(created.Proposal.Id);
        Assert.NotNull(reconciled);
        Assert.True(
            string.Equals(
                AgentProposalStates.Ready,
                reconciled!.Status,
                StringComparison.Ordinal),
            $"Expected ready after read-only reconciliation, got "
            + $"'{reconciled.Status}' ({reconciled.Error?.Code}: "
            + $"{reconciled.Error?.Message}).");
        Assert.Equal("release-analyst", reconciled.OpenClawAgentId);
        Assert.Equal(1, gateway.CreateCalls);
    }

    private const string OpenClawStubScript =
        """
        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
        import json
        import re
        import threading
        import time

        agents = {}
        lock = threading.Lock()

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, format, *args):
                pass

            def send_json(self, status, payload):
                body = json.dumps(payload).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                try:
                    self.wfile.write(body)
                except (BrokenPipeError, ConnectionResetError):
                    pass

            def do_GET(self):
                if self.path != "/agents":
                    self.send_json(404, {"error": "not_found"})
                    return
                with lock:
                    snapshot = list(agents.values())
                self.send_json(200, {"agents": snapshot})

            def do_POST(self):
                if self.path != "/agents/create":
                    self.send_json(404, {"error": "not_found"})
                    return
                length = int(self.headers.get("Content-Length", "0"))
                payload = json.loads(self.rfile.read(length) or b"{}")
                name = payload.get("name", "")
                agent_id = re.sub(r"[^a-z0-9_-]+", "-", name.strip().lower())
                agent_id = re.sub(r"^-+|-+$", "", agent_id)[:64] or "main"
                workspace = payload.get("workspace")
                created = {
                    "id": agent_id,
                    "agentId": agent_id,
                    "name": name,
                    "workspace": workspace
                }
                with lock:
                    agents[agent_id] = created

                # The mutation is committed before the response. Toxiproxy
                # blocks downstream bytes, recreating an uncertain dispatch.
                time.sleep(0.1)
                self.send_json(200, {
                    "ok": True,
                    "agentId": agent_id,
                    "name": name,
                    "workspace": workspace
                })

        ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
        """;
}

internal sealed class ToxiproxyHttpGatewayConnector(Uri endpoint)
    : IGatewayConnector, IDisposable
{
    private readonly HttpClient client = new()
    {
        BaseAddress = endpoint
    };
    private int createCalls;

    public int CreateCalls => Volatile.Read(ref createCalls);
    public Func<CancellationToken, Task>? BeforeCreateAsync { get; set; }
    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 => "testcontainers-toxiproxy";
    public string? DeviceId => "toxiproxy-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)
    {
        using var timeoutSource =
            CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var budget = timeout is null
            ? TimeSpan.FromSeconds(1)
            : TimeSpan.FromMilliseconds(Math.Min(
                timeout.Value.TotalMilliseconds,
                1000));
        timeoutSource.CancelAfter(budget);

        try
        {
            HttpResponseMessage response;
            switch (method)
            {
                case "agents.list":
                    response = await client.GetAsync(
                        "/agents",
                        timeoutSource.Token);
                    break;
                case "agents.create":
                    Interlocked.Increment(ref createCalls);
                    if (BeforeCreateAsync is not null)
                        await BeforeCreateAsync(timeoutSource.Token);
                    response = await client.PostAsJsonAsync(
                        "/agents/create",
                        JsonSerializer.SerializeToNode(parameters),
                        timeoutSource.Token);
                    break;
                default:
                    throw new OpenClawGatewayRpcException(
                        "METHOD_NOT_FOUND",
                        $"Unexpected integration-test method {method}.");
            }

            using (response)
            {
                response.EnsureSuccessStatusCode();
                return JsonNode.Parse(
                    await response.Content.ReadAsStringAsync(
                        timeoutSource.Token));
            }
        }
        catch (OperationCanceledException)
            when (!cancellationToken.IsCancellationRequested)
        {
            throw new OpenClawGatewayRpcException(
                "GATEWAY_TIMEOUT",
                "Toxiproxy blocked the OpenClaw response.",
                retryable: true);
        }
        catch (HttpRequestException)
        {
            throw new OpenClawGatewayRpcException(
                "GATEWAY_DISCONNECTED",
                "Toxiproxy interrupted the OpenClaw connection.",
                retryable: true);
        }
    }

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

    public void Dispose() => client.Dispose();
}