nexus/backend-tests/ToxiproxyAgentProvisioningIntegrationTests.cs

using System.Net.Http.Json;
using System.Text;
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;

    [Fact]
    public void Http_fixture_preserves_json_node_request_payload()
    {
        var payload = ToxiproxyHttpGatewayConnector.SerializeParameters(
            new JsonObject
            {
                ["name"] = "Release Analyst",
                ["workspace"] = "/tmp/workspace-release-analyst"
            });
        var json = JsonNode.Parse(payload)!.AsObject();

        Assert.Equal("Release Analyst", json["name"]!.GetValue<string>());
        Assert.Equal(
            "/tmp/workspace-release-analyst",
            json["workspace"]!.GetValue<string>());
    }

    [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)
            .WithNetworkAliases("toxiproxy")
            .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);

        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}). Last test inventory: "
            + $"{gateway.LastAgentInventory ?? "<none>"}");
        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
        import urllib.request

        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

                # Install the downstream fault only after the mutation is
                # committed. This distinguishes an uncertain response from a
                # request that never crossed the dispatch boundary.
                toxic = json.dumps({
                    "name": "delay-create-response",
                    "type": "latency",
                    "stream": "downstream",
                    "toxicity": 1.0,
                    "attributes": {"latency": 1500, "jitter": 0}
                }).encode("utf-8")
                request = urllib.request.Request(
                    "http://toxiproxy:8474/proxies/openclaw/toxics",
                    data=toxic,
                    headers={"Content-Type": "application/json"},
                    method="POST")
                with urllib.request.urlopen(request, timeout=2):
                    pass

                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;
    private string? lastAgentInventory;

    public int CreateCalls => Volatile.Read(ref createCalls);
    public string? LastAgentInventory => Volatile.Read(ref lastAgentInventory);
    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);
                    var body = SerializeParameters(parameters);
                    using var content = new StringContent(
                        body,
                        Encoding.UTF8,
                        "application/json");
                    response = await client.PostAsync(
                        "/agents/create",
                        content,
                        timeoutSource.Token);
                    break;
                }
                default:
                    throw new OpenClawGatewayRpcException(
                        "METHOD_NOT_FOUND",
                        $"Unexpected integration-test method {method}.");
            }

            using (response)
            {
                response.EnsureSuccessStatusCode();
                var payload = await response.Content.ReadAsStringAsync(
                    timeoutSource.Token);
                if (method == "agents.list")
                    Interlocked.Exchange(ref lastAgentInventory, payload);
                return JsonNode.Parse(payload);
            }
        }
        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)
        => [];

    internal static string SerializeParameters(object? parameters)
        => parameters is JsonNode jsonNode
            ? jsonNode.ToJsonString()
            : JsonSerializer.Serialize(
                parameters,
                parameters?.GetType() ?? typeof(object));

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