XFE.SpaceEngineers.AgentBridge

#nullable enable
using System.Collections.Concurrent;
using System.Diagnostics;
using System.IO.Pipes;
using System.Security.AccessControl;
using System.Security.Principal;
using System.Text;
using System.Text.Json;
using Newtonsoft.Json.Linq;
using XFE.SeAgent.Cli;
using XFE.SeAgent.Plugin.Protocol;
using XFE.SeAgent.Plugin.Transport;

namespace XFE.SeAgent.Tests;

internal static class Program
{
    private static readonly string Fixtures = Path.Combine(Path.GetTempPath(), "XfeSeAgentTests", Guid.NewGuid().ToString("N"));
    private static readonly List<string> Checks = [];
    private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web);

    private static async Task<int> Main()
    {
        Directory.CreateDirectory(Fixtures);
        try
        {
            TestProtocol(); TestEndpoints();
            await TestClientBoundariesAsync();
            await TestTransportAndCliAsync();
            Console.WriteLine($"PASS {Checks.Count} checks; fixtures: {Fixtures}");
            File.WriteAllText(Path.Combine(Fixtures, "results.json"), JsonSerializer.Serialize(new { passed = true, checks = Checks }, Json));
            return 0;
        }
        catch (Exception ex) { Console.Error.WriteLine(ex); return 1; }
    }
    private static void Check(bool condition, string name) { if (!condition) throw new InvalidOperationException("FAIL " + name); Checks.Add(name); Console.WriteLine("PASS " + name); }
    private static void Throws<T>(Action action, string name) where T : Exception { try { action(); } catch (T) { Check(true, name); return; } throw new InvalidOperationException("FAIL expected " + typeof(T).Name + ": " + name); }
    private static async Task ThrowsAsync<T>(Func<Task> action, string name) where T : Exception { try { await action(); } catch (T) { Check(true, name); return; } throw new InvalidOperationException("FAIL expected " + typeof(T).Name + ": " + name); }
    private static JsonElement Params(object value) => JsonSerializer.SerializeToElement(value);

    private static void TestProtocol()
    {
        var request = AgentProtocol.ParseRequest("{\"id\":\"test\",\"method\":\"pb.read\",\"params\":{\"entityId\":\"1234567890123456789\"},\"timeoutMs\":30000}");
        Check((string)request.Id! == "test" && request.Parameters["entityId"]!.Value<string>() == "1234567890123456789" && request.TimeoutMs == 30000, "protocol retains string block identifiers and timeout");
        Check(AgentProtocol.ParseRequest("{\"id\":1,\"method\":\"world.status\"}").TimeoutMs == AgentProtocol.DefaultTimeoutMs, "protocol applies default timeout and empty params");
        foreach (string invalid in new[] { "{}", "[]", "{broken", "{\"id\":1,\"id\":2,\"method\":\"x\"}", "{\"id\":1,\"method\":\"bad name\"}", "{\"id\":1,\"method\":\"x\",\"params\":[]}", "{\"id\":1,\"method\":\"x\",\"timeoutMs\":0}", "{\"id\":1,\"method\":\"x\",\"timeoutMs\":30001}", "{\"id\":1,\"method\":\"x\"} {}" })
            Throws<AgentRpcException>(() => AgentProtocol.ParseRequest(invalid), "protocol rejects invalid request " + invalid);
        Throws<AgentRpcException>(() => AgentProtocol.ParseRequest(new string('x', AgentProtocol.MaxRequestBytes + 1)), "protocol request byte limit enforced");
        byte[] encoded = AgentProtocol.SerializeResponse(AgentProtocol.Success(new JValue("id"), new JObject { ["text"] = "中文\n换行" }));
        Check(encoded[^1] == '\n' && encoded.Count(b => b == '\n') == 1 && !encoded.Take(3).SequenceEqual(new byte[] { 239, 187, 191 }), "response is one UTF-8 JSON line without BOM");
        Throws<AgentRpcException>(() => AgentProtocol.SerializeResponse(AgentProtocol.Success(new JValue(1), new JObject { ["large"] = new string('x', AgentProtocol.MaxResponseBytes) })), "response serialization is bounded");
        var error = AgentProtocol.Error(new JValue(1), new string('c', 100), new string('m', 4000));
        Check(error["error"]!["code"]!.Value<string>()!.Length == 64 && error["error"]!["message"]!.Value<string>()!.Length == 2048, "public error fields bounded");
    }

    private static void TestEndpoints()
    {
        string root = Path.Combine(Fixtures, "endpoints"); Directory.CreateDirectory(root);
        string path = Environment.ProcessPath!; var start = DateTimeOffset.UtcNow;
        var inspector = new FixtureInspector(new Dictionary<int, ProcessIdentity> { [11] = new(path, start), [12] = new(path, start), [13] = new(path, start), [14] = new(path, start) });
        void Write(int id, AgentEndpoint endpoint) => File.WriteAllText(Path.Combine(root, id + ".json"), JsonSerializer.Serialize(endpoint, Json));
        var valid = new AgentEndpoint("1.0.0", 1, "XFE.SE.Agent.11", 11, start, path, start);
        Write(11, valid); Write(12, valid with { Pid = 12, PipeName = "XFE.SE.Agent.12", StartTimeUtc = start.AddSeconds(-1) });
        Write(13, valid with { Pid = 13, PipeName = "XFE.SE.Agent.13", ExecutablePath = Path.Combine(Fixtures, "different.exe") });
        Write(14, valid with { Pid = 14, PipeName = "wrong-pipe" }); Write(15, valid with { Pid = 15, PipeName = "XFE.SE.Agent.15" });
        File.WriteAllText(Path.Combine(root, "invalid.json"), "{broken");
        var discovered = new EndpointDiscovery(inspector).Discover(root);
        Check(discovered.Count == 1 && discovered[0] == valid, "discovery rejects stale start times, executable mismatch, pipe mismatch and exited processes");
        using var self = Process.GetCurrentProcess();
        var actual = new LocalProcessInspector().Inspect(self.Id);
        Check(actual?.StartTimeUtc.UtcTicks == new DateTimeOffset(self.StartTime.ToUniversalTime()).UtcTicks && actual.ExecutablePath == self.MainModule!.FileName, "real local process inspection verifies start time and executable path");
    }

    private static async Task TestClientBoundariesAsync()
    {
        var client = new AgentPipeClient();
        await ThrowsAsync<AgentClientException>(() => client.CallAsync("missing-" + Guid.NewGuid().ToString("N"), "world.status", Params(new { }), 30), "client missing-pipe connect timeout is bounded");
        using (var cancel = new CancellationTokenSource(30))
            await ThrowsAsync<OperationCanceledException>(() => client.CallAsync("missing-" + Guid.NewGuid().ToString("N"), "world.status", Params(new { }), 1000, cancel.Token), "caller cancellation is distinguished from timeout");
        await ThrowsAsync<AgentClientException>(() => client.CallAsync("fixture", "pb.deploy", Params(new { source = new string('x', AgentPipeClient.MaximumRequestBytes) }), 1000), "oversized request rejected before pipe connection");
        foreach (var response in new[] { "not json\n", "{\"id\":\"wrong\",\"result\":{}}\n", "{\"id\":\"$ID\",\"result\":{},\"error\":{}}\n", "{\"id\":\"$ID\",\"result\":{}}" })
        {
            string name = "xfe-client-test-" + Guid.NewGuid().ToString("N");
            var serve = ServeRawOnceAsync(name, request => Encoding.UTF8.GetBytes(response.Replace("$ID", request.GetProperty("id").GetString())));
            await ThrowsAsync<AgentClientException>(() => client.CallAsync(name, "world.status", Params(new { }), 1000), "client rejects malformed, mismatched or incomplete response: " + response);
            await serve;
        }
    }

    private static async Task ServeRawOnceAsync(string pipeName, Func<JsonElement, byte[]> reply)
    {
        await using var pipe = new NamedPipeServerStream(pipeName, PipeDirection.InOut, 1, PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
        await pipe.WaitForConnectionAsync();
        using var reader = new StreamReader(pipe, new UTF8Encoding(false), false, 1024, leaveOpen: true);
        using var document = JsonDocument.Parse((await reader.ReadLineAsync())!);
        await pipe.WriteAsync(reply(document.RootElement)); await pipe.FlushAsync();
    }

    private static async Task TestTransportAndCliAsync()
    {
        var executed = new ConcurrentQueue<(string Method, JObject Parameters, int Thread)>();
        using var server = new AgentServer((method, parameters) =>
        {
            executed.Enqueue((method, (JObject)parameters.DeepClone(), Environment.CurrentManagedThreadId));
            if (method == "fixture.error") throw new AgentRpcException("fixture_error", "expected game error");
            if (method == "fixture.large") return new JObject { ["payload"] = new string('x', AgentProtocol.MaxResponseBytes) };
            return new JObject { ["method"] = method, ["parameters"] = parameters.DeepClone() };
        }, _ => { });
        server.Start();
        using var pump = new PumpThread(server);
        using (var aclClient = new NamedPipeClientStream(".", server.PipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
        {
            await aclClient.ConnectAsync(1000);
            var security = aclClient.GetAccessControl();
            var rules = security.GetAccessRules(true, true, typeof(SecurityIdentifier)).Cast<PipeAccessRule>().ToArray();
            using var identity = WindowsIdentity.GetCurrent();
            var network = new SecurityIdentifier(WellKnownSidType.NetworkSid, null);
            Check(security.AreAccessRulesProtected && rules.Where(rule => rule.AccessControlType == AccessControlType.Allow).All(rule => rule.IdentityReference.Equals(identity.User)) && rules.Any(rule => rule.AccessControlType == AccessControlType.Deny && rule.IdentityReference.Equals(network)), "real pipe ACL permits only current user and denies network access");
        }
        var client = new AgentPipeClient();
        var held = client.CallAsync(server.PipeName, "world.status", Params(new { }), 2000);
        await Task.Delay(80);
        Check(executed.IsEmpty && !held.IsCompleted, "pipe worker queues work without touching game callback before Pump");
        pump.Enabled = true;
        var answer = await held;
        Check(!answer.IsError && executed.All(item => item.Thread == pump.ThreadId), "only dedicated game-pump thread executes queued operations");
        var concurrent = await Task.WhenAll(Enumerable.Range(0, 4).Select(index => client.CallAsync(server.PipeName, "blocks.get", Params(new { entityId = index.ToString() }), 2000)));
        Check(concurrent.All(reply => !reply.IsError), "four concurrent clients receive independent correlated responses");
        var serverError = await client.CallAsync(server.PipeName, "fixture.error", Params(new { }), 2000);
        Check(serverError.IsError && serverError.Envelope.GetProperty("error").GetProperty("code").GetString() == "fixture_error", "game RPC errors retain structured code and message");
        var oversized = await client.CallAsync(server.PipeName, "fixture.large", Params(new { }), 2000);
        Check(oversized.IsError && oversized.Envelope.GetProperty("error").GetProperty("code").GetString() == "response_too_large", "oversized game result becomes bounded protocol error");
        await TestCliAsync(server.PipeName, executed);
        using (var raw = new NamedPipeClientStream(".", server.PipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
        {
            await raw.ConnectAsync(1000);
            await raw.WriteAsync(Encoding.UTF8.GetBytes(new string('x', AgentProtocol.MaxRequestBytes + 1) + "\n"));
            await raw.FlushAsync();
            using var reader = new StreamReader(raw, Encoding.UTF8, false, 1024, true);
            var response = JObject.Parse((await reader.ReadLineAsync())!);
            Check(response["error"]!["code"]!.Value<string>() == "request_too_large", "actual pipe transport rejects oversized request before execution");
        }
        pump.Enabled = false; await Task.Delay(30);
        int before = executed.Count;
        try
        {
            var expired = await client.CallAsync(server.PipeName, "pb.deploy", Params(new { }), 50);
            Check(expired.IsError && expired.Envelope.GetProperty("error").GetProperty("code").GetString() == "timeout", "request timeout propagates while game pump is paused");
        }
        catch (AgentClientException error) when (error.Code == "timeout") { Check(true, "request timeout propagates while game pump is paused"); }
        await Task.Delay(30); pump.Enabled = true; await Task.Delay(50);
        Check(executed.Count == before, "expired request cannot execute later after Pump resumes");
        pump.Enabled = false; await Task.Delay(20);
        using (var raw = new NamedPipeClientStream(".", server.PipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
        {
            await raw.ConnectAsync(1000); await raw.WriteAsync(Encoding.UTF8.GetBytes("{\"id\":\"disconnect\",\"method\":\"pb.deploy\",\"params\":{},\"timeoutMs\":1000}\n")); await raw.FlushAsync();
        }
        await Task.Delay(40); pump.Enabled = true; await Task.Delay(40);
        Check(executed.Count == before, "disconnected queued mutation is discarded before game execution");
        pump.Enabled = false; await Task.Delay(20);
        var stopped = client.CallAsync(server.PipeName, "pb.deploy", Params(new { }), 1000); await Task.Delay(40); server.Dispose();
        try { await stopped; } catch (AgentClientException) { }
        pump.Enabled = true; await Task.Delay(40);
        Check(executed.Count == before, "disposing server cancels pending mutations before further Pump");
    }

    private static async Task TestCliAsync(string pipe, ConcurrentQueue<(string Method, JObject Parameters, int Thread)> executed)
    {
        var cli = new CliApplication();
        async Task<(int Exit, JsonElement Json, string Error)> Run(params string[] args)
        {
            using var output = new StringWriter(); using var error = new StringWriter();
            int code = await cli.RunAsync(args, output, error);
            using var json = JsonDocument.Parse(output.ToString());
            return (code, json.RootElement.Clone(), error.ToString());
        }
        var help = await Run("--help"); Check(help.Exit == 0 && help.Json.GetProperty("commands").GetArrayLength() == 4 && help.Error.Length == 0, "CLI help is one valid JSON document with no stdout diagnostics");
        string parameters = Path.Combine(Fixtures, "parameters.json"); File.WriteAllText(parameters, "{\"entityId\":\"1234567890123456789\",\"argument\":\"中文 参数\"}");
        var called = await Run("call", "pb.run", "--params", "@" + parameters, "--pipe", pipe);
        Check(called.Exit == 0 && called.Error.Length == 0 && called.Json.GetProperty("result").GetProperty("parameters").GetProperty("argument").GetString() == "中文 参数", "CLI call reads @JSON file and preserves Unicode parameters");
        string script = Path.Combine(Fixtures, "script.cs"); const string source = "// 中文脚本\npublic void Main() { Echo(\"hello\"); }"; File.WriteAllText(script, source);
        string hash = new('a', 64);
        var deployed = await Run("deploy", "--block", "1234567890123456789", "--file", script, "--expected-hash", hash, "--pipe", pipe);
        var deploy = executed.Last(item => item.Method == "pb.deploy");
        Check(deployed.Exit == 0 && deploy.Parameters["source"]!.Value<string>() == source && deploy.Parameters["expectedSha256"]!.Value<string>() == hash && deploy.Parameters["entityId"]!.Type == JTokenType.String, "CLI deploy transmits exact source and mandatory optimistic hash with string entity ID");
        int count = executed.Count;
        var refused = await Run("deploy", "--block", "123", "--file", script, "--pipe", pipe);
        Check(refused.Exit == 2 && refused.Json.TryGetProperty("error", out _) && executed.Count == count, "CLI deploy without expected hash fails before sending mutation");
        var failure = await Run("call", "fixture.error", "--pipe", pipe);
        Check(failure.Exit == 4 && failure.Json.GetProperty("error").GetProperty("code").GetString() == "fixture_error" && failure.Error.Length > 0, "CLI RPC failure remains JSON on stdout, diagnostic on stderr and nonzero exit");
        string jsonl = Path.Combine(Fixtures, "telemetry.jsonl");
        var watched = await Run("watch", "--block", "123,-456", "--seconds", "0.14", "--interval", "0.05", "--out", jsonl, "--pipe", pipe);
        string[] lines = File.ReadAllLines(jsonl);
        Check(watched.Exit == 0 && lines.Length >= 2 && lines.Length == watched.Json.GetProperty("samples").GetInt32(), "CLI watch writes complete JSONL samples and one stdout summary");
        using var sample = JsonDocument.Parse(lines[0]);
        Check(sample.RootElement.GetProperty("response").GetProperty("result").GetProperty("parameters").GetProperty("entityIds").EnumerateArray().Select(value => value.GetString()).SequenceEqual(new[] { "123", "-456" }), "watch preserves positive and negative Int64 block IDs as strings");
        var bad = await Run("call", "world.status", "--timeout", "31", "--pipe", pipe);
        Check(bad.Exit == 2 && bad.Error.Length > 0, "CLI validates request timeout against server maximum");
        var conflict = await Run("call", "world.status", "--pipe", pipe, "--pid", "1");
        Check(conflict.Exit == 2, "CLI rejects ambiguous simultaneous pipe and PID selectors");
        var invalidPid = await Run("discover", "--pid", "invalid", "--endpoint-directory", Path.Combine(Fixtures, "empty"));
        Check(invalidPid.Exit == 2, "discover validates PID even when no endpoints exist");
        var info = new ProcessStartInfo(Path.Combine(AppContext.BaseDirectory, "xfe-se.exe")) { UseShellExecute = false, CreateNoWindow = true, RedirectStandardOutput = true, RedirectStandardError = true };
        foreach (string argument in new[] { "call", "world.status", "--pipe", pipe, "--params", "{\"argument\":\"真实 CLI 进程\"}" }) info.ArgumentList.Add(argument);
        using var child = Process.Start(info)!;
        var stdout = child.StandardOutput.ReadToEndAsync(); var stderr = child.StandardError.ReadToEndAsync();
        await child.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(5));
        using var childJson = JsonDocument.Parse(await stdout);
        Check(child.ExitCode == 0 && (await stderr).Length == 0 && childJson.RootElement.GetProperty("result").GetProperty("parameters").GetProperty("argument").GetString() == "真实 CLI 进程", "built CLI executable performs real pipe call with pure JSON stdout");
    }

    private sealed class FixtureInspector(Dictionary<int, ProcessIdentity> processes) : IProcessInspector
    { public ProcessIdentity? Inspect(int pid) => processes.GetValueOrDefault(pid); }

    private sealed class PumpThread : IDisposable
    {
        private readonly Thread thread;
        private volatile bool enabled, stopped;
        private Exception? failure;
        public int ThreadId => thread.ManagedThreadId;
        public bool Enabled { get => enabled; set => enabled = value; }
        public PumpThread(AgentServer server)
        {
            thread = new Thread(() =>
            {
                try { while (!stopped) { if (enabled) server.Pump(); Thread.Sleep(1); } }
                catch (Exception ex) { failure = ex; }
            }) { IsBackground = true, Name = "owned-game-pump-fixture" };
            thread.Start();
        }
        public void Dispose() { stopped = true; thread.Join(2000); if (failure != null) throw new InvalidOperationException("Game pump fixture failed", failure); }
    }
}