perf(tests): isolate network E2E and await QUIC cleanup

backend/agent/src/MoShiro.Agent/AgentApp.cs

@@ -11,7 +11,7 @@ using MoShiro.Protocol;
 namespace MoShiro.Agent;
 
 /// <summary>Agent 的组装点：数据面、点对点监听、与 Backend 的链路都在这里接起来。没有全局状态，测试可以在同一进程里启动多个。</summary>
-public sealed class AgentApp : IDisposable
+public sealed class AgentApp : IDisposable, IAsyncDisposable
 {
     public const string Version = MoShiro.Protocol.BuildInfo.Version;
 
@@ -40,6 +40,7 @@ public sealed class AgentApp : IDisposable
     private readonly TaskManager _tasks;
     private readonly TestRunner _tests;
     private readonly FleetLink _link;
+    private int _disposed;
 
     /// <param name="dataDirectory">配置目录（容器里的 <c>/etc/moshiro</c>）：宿主机更新器和节点在这里交换清单、请求和状态。</param>
     public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
@@ -113,10 +114,26 @@ public sealed class AgentApp : IDisposable
 
     public void Dispose()
     {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
         _datagrams.Dispose();
         _ownedRestart?.Dispose();
     }
 
+    public async ValueTask DisposeAsync()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        await _datagrams.DisposeAsync().ConfigureAwait(false);
+        _ownedRestart?.Dispose();
+    }
+
     internal StreamRegistry Streams => _streams;
 
     public IReadOnlyList<int> QuicPorts => _datagrams.BoundPorts;

backend/agent/src/MoShiro.Agent/Program.cs

@@ -43,7 +43,7 @@ catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidO
 using var restartScope = restart;
 using (identity)
 {
-    using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log, restart);
+    await using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log, restart);
     await ServiceShell.RunAsync(app.RunAsync, log, restart, time);
 }
 

backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs

@@ -196,7 +196,7 @@ public sealed class AgentConnection : IDisposable
         var remote = (IPEndPoint)info.Remote!;
         IReadOnlyList<int> quicPorts = Endpoint.EffectiveQuicPorts;
         int quicPort = quicPorts.Count == 0 ? 7443 : quicPorts[0];
-        using var datagrams = new QuicDatagramTransport(
+        await using var datagrams = new QuicDatagramTransport(
             data,
             _env.Clock,
             new DataPlaneCounters(),

backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs

@@ -10,7 +10,7 @@ namespace MoShiro.Measurement;
 /// open one QUIC connection from a newly selected ephemeral source port. Measurement frames are sent
 /// only with MsQuic DATAGRAM calls; the existing TLS peer connection continues to carry session control.
 /// </summary>
-public sealed class QuicDatagramTransport : IDisposable
+public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
 {
     private readonly MeasurementDatagramRouter _router;
     private readonly INodeLog _log;
@@ -166,6 +166,16 @@ public sealed class QuicDatagramTransport : IDisposable
         _ = DisposeNativeResourcesAsync();
     }
 
+    public async ValueTask DisposeAsync()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        await DisposeNativeResourcesAsync().ConfigureAwait(false);
+    }
+
     private async Task DisposeNativeResourcesAsync()
     {
         try

backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs

@@ -0,0 +1,78 @@
+using System.Net;
+using System.Net.Sockets;
+using MoShiro.Measurement;
+using MoShiro.Protocol;
+using MoShiro.TestSupport;
+
+namespace MoShiro.Backend.Tests;
+
+/// <summary>数据面协议拒绝错误首帧、伪造包、重放包和垃圾数据。</summary>
+[Trait("Category", "Slow")]
+public sealed class AgentDataPlaneTests : IClassFixture<BackendFixture>
+{
+    private readonly BackendFixture _backend;
+
+    public AgentDataPlaneTests(BackendFixture backend)
+    {
+        _backend = backend;
+    }
+
+    private async Task<(AgentHarness Agent, ProbeSimulator Probe, LineTask Udp)> LineAsync()
+    {
+        AgentHarness agent = await AgentHarness.StartAsync(_backend);
+        await agent.OverrideLineTasksAsync(rateHz: 30);
+        ProbeSimulator probe = await agent.ConnectProbeAsync();
+        LineTaskSet set = await probe.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0]));
+        return (agent, probe, set.Tasks.Single(t => t.Kind == "udp_echo"));
+    }
+
+    [Fact]
+    public async Task TcpDataPlane_ClosesConnectionsWithoutAValidFirstFrame()
+    {
+        (AgentHarness agent, ProbeSimulator probe, _) = await LineAsync();
+        await using (agent)
+        await using (probe)
+        {
+            using var client = new TcpClient();
+            await client.ConnectAsync(IPAddress.Loopback, agent.App.TcpPorts[0]);
+            await client.GetStream().WriteAsync(new byte[TcpWire.AuthBytes]); // 全零：magic 不对
+            byte[] buffer = new byte[16];
+            Assert.Equal(0, await ReadOrResetAsync(client.GetStream(), buffer)); // 连接被关闭：没有任何字节（Windows 上可能是 RST）
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid >= 1));
+        }
+    }
+
+    [Fact]
+    public async Task DataPlane_DropsForgedReplayedAndGarbagePackets_WithoutAnswering()
+    {
+        (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync();
+        await using (agent)
+        await using (probe)
+        {
+            long before = agent.App.Counters.DroppedInvalid;
+            await probe.SendRawUdpAsync([1, 2, 3]);
+            await probe.SendRawUdpAsync(new byte[64]);
+            await probe.SendForgedEchoAsync(udp, 1);
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid - before >= 3), because: "garbage and forged packets are counted");
+
+            // 重放：同一个序号发两次，只有第一次被接受
+            await probe.SendEchoAsync(udp, 500);
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.PacketsOut >= 1));
+            long dropped = agent.App.Counters.DroppedInvalid;
+            await probe.SendEchoAsync(udp, 500);
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid > dropped), because: "a replayed packet is dropped");
+        }
+    }
+
+    private static async Task<int> ReadOrResetAsync(Stream stream, byte[] buffer)
+    {
+        try
+        {
+            return await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10));
+        }
+        catch (IOException)
+        {
+            return 0;
+        }
+    }
+}

backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs

@@ -6,6 +6,8 @@ using MoShiro.TestSupport;
 namespace MoShiro.Backend.Tests;
 
 /// <summary>真实的后端 + 真实的 Agent（进程内）+ 假 Probe：覆盖 Agent 与 Backend 的链路、Probe 与 Agent 的点对点会话和数据面。</summary>
+// 使用真实 Agent/QUIC/TCP 会话的端到端验证放在慢测集合，避免和大量 API/单元测试争抢本机套接字。
+[Trait("Category", "Slow")]
 public sealed class AgentEndToEndTests : IClassFixture<BackendFixture>
 {
     private readonly BackendFixture _backend;
@@ -52,20 +54,6 @@ public sealed class AgentEndToEndTests : IClassFixture<BackendFixture>
         Assert.Equal("stopping", final["runtime_status"]!.GetValue<string>());
     }
 
-    [Fact]
-    public async Task LocalLimits_RejectTasksOutsideTheAllowedPorts_AndTheReasonReachesTheAdmin()
-    {
-        // 默认模板的端口（7443/7444）不在这个 Agent 实际监听的端口里：任务被拒绝，原因是 policy_denied
-        await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
-        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == Defaults.TaskCount);
-        JsonNode rejected = (await AgentViewAsync(agent))["rejected"]!;
-        Assert.All(rejected.AsArray(), item => Assert.Equal("policy_denied", item!["reason"]!.GetValue<string>()));
-
-        // 覆盖成实际端口之后恢复
-        await agent.OverrideLineTasksAsync();
-        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 0);
-    }
-
     [Fact]
     public async Task UdpEcho_EndToEnd_FragmentsAreMergedWithAgentTimestamps_AndStoredByTheBackend()
     {
@@ -108,57 +96,6 @@ public sealed class AgentEndToEndTests : IClassFixture<BackendFixture>
         }
     }
 
-    /// <summary>读到 0 字节（对端关闭）或者连接被重置，都算「没收到任何东西」；收到数据返回字节数。</summary>
-    internal static async Task<int> ReadOrResetAsync(Stream stream, byte[] buffer)
-    {
-        try
-        {
-            return await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10));
-        }
-        catch (IOException)
-        {
-            return 0;
-        }
-    }
-
-    [Fact]
-    public async Task TcpDataPlane_ClosesConnectionsWithoutAValidFirstFrame()
-    {
-        (AgentHarness agent, ProbeSimulator probe, _) = await LineAsync();
-        await using (agent)
-        await using (probe)
-        {
-            using var client = new System.Net.Sockets.TcpClient();
-            await client.ConnectAsync(System.Net.IPAddress.Loopback, agent.App.TcpPorts[0]);
-            await client.GetStream().WriteAsync(new byte[TcpWire.AuthBytes]); // 全零：magic 不对
-            byte[] buffer = new byte[16];
-            Assert.Equal(0, await ReadOrResetAsync(client.GetStream(), buffer)); // 连接被关闭：没有任何字节（Windows 上可能是 RST）
-            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid >= 1));
-        }
-    }
-
-    [Fact]
-    public async Task DataPlane_DropsForgedReplayedAndGarbagePackets_WithoutAnswering()
-    {
-        (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync();
-        await using (agent)
-        await using (probe)
-        {
-            long before = agent.App.Counters.DroppedInvalid;
-            await probe.SendRawUdpAsync([1, 2, 3]);
-            await probe.SendRawUdpAsync(new byte[64]);
-            await probe.SendForgedEchoAsync(udp, 1);
-            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid - before >= 3), because: "garbage and forged packets are counted");
-
-            // 重放：同一个序号发两次，只有第一次被接受
-            await probe.SendEchoAsync(udp, 500);
-            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.PacketsOut >= 1));
-            long dropped = agent.App.Counters.DroppedInvalid;
-            await probe.SendEchoAsync(udp, 500);
-            await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid > dropped), because: "a replayed packet is dropped");
-        }
-    }
-
     [Fact]
     public async Task ARestartedProbe_NeverInheritsTheAgentRecordsOfItsPreviousSession()
     {
@@ -253,95 +190,6 @@ public sealed class AgentEndToEndTests : IClassFixture<BackendFixture>
         }
     }
 
-    [Fact]
-    public async Task LineTaskSet_FollowsTheDesiredState_AndIsPushedToTheProbe()
-    {
-        (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync();
-        await using (agent)
-        await using (probe)
-        {
-            JsonNode tasks = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/tasks");
-            JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue<string>() == agent.Id && t["kind"]!.GetValue<string>() == "udp_echo")!;
-            using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue<string>(), BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json"));
-            disabled.EnsureSuccessStatusCode();
-
-            LineTaskSet next = await probe.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "udp_echo") && s.Tasks.Count >= 0 && s.Version > 0);
-            Assert.DoesNotContain(next.Tasks, t => t.Id == udp.Id);
-        }
-    }
-
-    // 慢的端到端测试（真实时间的阶梯、要等测量块出来）：CI 里单独一批跑（Category=Slow），不拦着出镜像，见 ci/verify-backend.sh
-    [Fact]
-    [Trait("Category", "Slow")]
-    public async Task Hy2Ladder_EndToEnd_AgentSendsTheTrain_ProbeReportsReceipts_AndTheTestSucceeds()
-    {
-        await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
-        await agent.OverrideLineTasksAsync();
-        await using ProbeSimulator probe = await agent.ConnectProbeAsync();
-        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>());
-
-        HttpResponseMessage? accepted = null;
-        string? startFailure = null;
-        for (int attempt = 0; attempt < 5; attempt++)
-        {
-            accepted = await _backend.Admin.PostAsync($"/api/admin/lines/{agent.Id}~{probe.ProbeId}/tests", BackendFixture.Json(new JsonObject { ["plan"] = "hy2_ladder", ["params"] = new JsonObject { ["size"] = 1000, ["duration_s"] = 5 } }));
-            if (accepted.StatusCode == System.Net.HttpStatusCode.Accepted)
-            {
-                break;
-            }
-
-            startFailure = await accepted.Content.ReadAsStringAsync();
-            string? code = JsonNode.Parse(startFailure)?["code"]?.GetValue<string>();
-            accepted.Dispose();
-            accepted = null;
-            if (code != "probe_offline" || attempt == 4)
-            {
-                Assert.Fail($"The test request was rejected: {startFailure}");
-            }
-
-            await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>(), TimeSpan.FromSeconds(15), "the probe should reconnect after a transient disconnect");
-        }
-
-        HttpResponseMessage acceptedResponse = accepted ?? throw new InvalidOperationException("The test request was not accepted.");
-        using (acceptedResponse)
-        {
-            string testId = JsonNode.Parse(await acceptedResponse.Content.ReadAsStringAsync())!["id"]!.GetValue<string>();
-
-        JsonNode test = null!;
-        await BackendFixture.EventuallyAsync(
-            async () =>
-            {
-                test = await _backend.GetJsonAsync(_backend.Admin, "/api/public/tests/" + testId);
-                return test["status"]!.GetValue<string>() is "succeeded" or "failed" or "aborted";
-            },
-            TimeSpan.FromSeconds(90));
-        Assert.True(test["status"]!.GetValue<string>() == "succeeded", test.ToJsonString());
-        Assert.Equal(4, probe.Phases.Count);
-        Assert.All(probe.Phases, phase => Assert.Equal("udp_train", phase.Kind));
-
-        // 单块最多 50000 条记录：50 和 100 Mbit/s 阶段（5 s、1000 B 的包）里 100 Mbit/s 的那一段有 6 万多条记录，被 Probe 切成两个列块
-        Assert.True(agent.App.Counters.Fragments > 4);
-        await BackendFixture.EventuallyAsync(
-            async () => (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Count(c => c!["kind"]!.GetValue<string>() == "udp_train") == agent.App.Counters.Fragments,
-            because: (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.Id}")).ToJsonString() + $" fragments={agent.App.Counters.Fragments} discarded={agent.App.Counters.FragmentsDiscarded} pending={agent.App.Link.Outbox.PendingCount}/{agent.App.Link.Outbox.PendingBytes} dropped={agent.App.Link.Outbox.Dropped} link={agent.App.Link.IsConnected}");
-        }
-    }
-
-    [Fact]
-    public async Task Plan_UsingTcpBulk_IsAcceptedByARealAgentAndStartsItsFirstPhase()
-    {
-        await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
-        await using ProbeSimulator probe = await agent.ConnectProbeAsync();
-        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>());
-
-        // reality_session 需要 connect、tcp_bulk、hold：真实的 Agent 都实现了
-        using HttpResponseMessage response = await _backend.Admin.PostAsync($"/api/admin/lines/{agent.Id}~{probe.ProbeId}/tests", BackendFixture.Json(new JsonObject { ["plan"] = "reality_session" }));
-        Assert.Equal(System.Net.HttpStatusCode.Accepted, response.StatusCode);
-        await BackendFixture.EventuallyAsync(() => Task.FromResult(!probe.Phases.IsEmpty), because: "the first phase should start");
-        Assert.True(probe.Phases.TryPeek(out var first));
-        Assert.Equal("connect", first.Kind);
-    }
-
     [Fact]
     public async Task Test_IsRefusedWhenNoProbeIsConnected_AndWhenAPhaseExceedsTheLocalLimits()
     {

backend/tests/MoShiro.Backend.Tests/AgentHarness.cs

@@ -46,7 +46,7 @@ public sealed class AgentHarness : IAsyncDisposable
     {
         string directory = Path.Combine(Path.GetTempPath(), "moshiro-agent-" + Guid.NewGuid().ToString("N"));
         string token = pin?.Code ?? await backend.CreateJoinTokenAsync();
-        using var http = new HttpClient();
+        using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
         Result<EnrollmentResult> enrolled = await EnrollmentClient.EnrollAsync(
             http, backend.BaseAddress, token, name ?? backend.UniqueName("agent"), directory, AgentApp.Version, null, TimeProvider.System, CancellationToken.None, pin?.Genesis, pin?.Controllers);
         Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString());
@@ -62,7 +62,7 @@ public sealed class AgentHarness : IAsyncDisposable
         };
         configure?.Invoke(config);
         LoadedNodeIdentity identity = LoadedNodeIdentity.Load(directory);
-        var app = new AgentApp(config, identity, directory, TimeProvider.System, new ConsoleNodeLog(TimeProvider.System, Console.Out));
+        var app = new AgentApp(config, identity, directory, TimeProvider.System, NullNodeLog.Instance);
         var harness = new AgentHarness(backend, directory, identity, app);
         await app.Ready.WaitAsync(TimeSpan.FromSeconds(10));
         await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend");
@@ -137,7 +137,7 @@ public sealed class AgentHarness : IAsyncDisposable
             // 停止中
         }
 
-        App.Dispose();
+        await App.DisposeAsync();
         _identity.Dispose();
         _stop.Dispose();
         try

backend/tests/MoShiro.Backend.Tests/AgentLimitsTests.cs

@@ -0,0 +1,31 @@
+using System.Text.Json.Nodes;
+
+namespace MoShiro.Backend.Tests;
+
+/// <summary>Agent local policy validation has its own Backend fixture so task overrides from line-session tests cannot leak into it.</summary>
+[Trait("Category", "Slow")]
+public sealed class AgentLimitsTests : IClassFixture<BackendFixture>
+{
+    private readonly BackendFixture _backend;
+
+    public AgentLimitsTests(BackendFixture backend)
+    {
+        _backend = backend;
+    }
+
+    private async Task<JsonNode> AgentViewAsync(AgentHarness agent) => await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id);
+
+    [Fact]
+    public async Task LocalLimits_RejectTasksOutsideTheAllowedPorts_AndTheReasonReachesTheAdmin()
+    {
+        // The default task ports (7443/7444) do not match this Agent's actual listeners, so policy rejects them.
+        await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
+        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == Defaults.TaskCount);
+        JsonNode rejected = (await AgentViewAsync(agent))["rejected"]!;
+        Assert.All(rejected.AsArray(), item => Assert.Equal("policy_denied", item!["reason"]!.GetValue<string>()));
+
+        // Override the task ports to match the actual listeners and verify recovery.
+        await agent.OverrideLineTasksAsync();
+        await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 0);
+    }
+}

backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs

@@ -0,0 +1,36 @@
+using System.Text.Json.Nodes;
+using MoShiro.Measurement;
+using MoShiro.Protocol;
+using MoShiro.TestSupport;
+
+namespace MoShiro.Backend.Tests;
+
+/// <summary>Exercises the complete Backend → Agent → Probe desired task update with an isolated Backend fixture.</summary>
+[Trait("Category", "Slow")]
+public sealed class AgentTaskSetEndToEndTests : IClassFixture<BackendFixture>
+{
+    private readonly BackendFixture _backend;
+
+    public AgentTaskSetEndToEndTests(BackendFixture backend)
+    {
+        _backend = backend;
+    }
+
+    [Fact]
+    public async Task LineTaskSet_FollowsTheDesiredState_AndIsPushedToTheProbe()
+    {
+        await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
+        await agent.OverrideLineTasksAsync(rateHz: 30);
+        await using ProbeSimulator probe = await agent.ConnectProbeAsync();
+        LineTask udp = (await probe.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0])))
+            .Tasks.Single(t => t.Kind == "udp_echo");
+
+        JsonNode tasks = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/tasks");
+        JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue<string>() == agent.Id && t["kind"]!.GetValue<string>() == "udp_echo")!;
+        using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue<string>(), BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json"));
+        disabled.EnsureSuccessStatusCode();
+
+        LineTaskSet next = await probe.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "udp_echo") && s.Version > 0);
+        Assert.DoesNotContain(next.Tasks, t => t.Id == udp.Id);
+    }
+}

backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs

@@ -20,7 +20,7 @@ public sealed class AotSmokeTests
     {
         Assert.SkipUnless(Url is not null, "set MOSHIRO_AOT_URL to the address of a running NativeAOT build to run this test");
         var baseAddress = new Uri(Url!);
-        using var http = new HttpClient { BaseAddress = baseAddress };
+        using var http = new HttpClient { BaseAddress = baseAddress, Timeout = TimeSpan.FromSeconds(15) };
 
         // 管理员：登录 → 令牌
         using HttpResponseMessage login = await http.PostAsync("/api/admin/auth/login", BackendFixture.Json(new JsonObject { ["username"] = BackendFixture.AdminUser, ["password"] = BackendFixture.AdminPassword }));
@@ -41,7 +41,7 @@ public sealed class AotSmokeTests
         using HttpResponseMessage probeJoin = await http.PostAsync("/api/admin/join-tokens", BackendFixture.Json(new JsonObject { ["role"] = "probe", ["max_uses"] = 1 }));
         probeJoin.EnsureSuccessStatusCode();
         string probeToken = JsonNode.Parse(await probeJoin.Content.ReadAsStringAsync())!["plaintext"]!.GetValue<string>();
-        using var probeHttp = new HttpClient();
+        using var probeHttp = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
         Fleet.Core.Result<Fleet.Enrollment.Client.EnrollmentResult> probeEnrolled = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync(
             probeHttp, baseAddress, probeToken, "aot-probe-" + Guid.NewGuid().ToString("N")[..6], Path.Combine(Path.GetTempPath(), "moshiro-aot-probe-" + Guid.NewGuid().ToString("N")), "0.0.0", null, TimeProvider.System, default);
         Assert.True(probeEnrolled.IsSuccess, probeEnrolled.IsSuccess ? "" : probeEnrolled.Error.ToString());
@@ -51,7 +51,7 @@ public sealed class AotSmokeTests
         using HttpResponseMessage joinResponse = await http.PostAsync("/api/admin/join-tokens", BackendFixture.Json(new JsonObject { ["max_uses"] = 1 }));
         joinResponse.EnsureSuccessStatusCode();
         string joinToken = JsonNode.Parse(await joinResponse.Content.ReadAsStringAsync())!["plaintext"]!.GetValue<string>();
-        using var anonymous = new HttpClient { BaseAddress = baseAddress };
+        using var anonymous = new HttpClient { BaseAddress = baseAddress, Timeout = TimeSpan.FromSeconds(15) };
         await using AgentSimulator agent = await AgentSimulator.EnrollAsync(anonymous, joinToken, "aot-" + Guid.NewGuid().ToString("N")[..6]);
         agent.ProbeForChunks = new ProbeId(probeId);
         string lineId = $"{agent.NodeId.Value}~{probeId}";

backend/tests/MoShiro.Backend.Tests/BackendFixture.cs

@@ -65,7 +65,7 @@ public class BackendFixture : IAsyncLifetime
     {
         string token = await CreateJoinTokenAsync(role: "probe");
         string directory = Path.Combine(Path.GetTempPath(), "moshiro-probe-id-" + Guid.NewGuid().ToString("N"));
-        using var http = new HttpClient();
+        using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
         Fleet.Core.Result<Fleet.Enrollment.Client.EnrollmentResult> enrolled = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync(
             http, BaseAddress, token, name ?? UniqueName("probe"), directory, "0.0.0-test", null, TimeProvider.System, default);
         Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString());
@@ -179,7 +179,7 @@ public class BackendFixture : IAsyncLifetime
         return (client, passkey, name);
     }
 
-    public HttpClient NewClient() => new() { BaseAddress = BaseAddress };
+    public HttpClient NewClient() => new() { BaseAddress = BaseAddress, Timeout = TimeSpan.FromSeconds(15) };
 
     public string UniqueName(string prefix) => $"{prefix}-{Interlocked.Increment(ref _counter)}";
 

backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs

@@ -376,7 +376,7 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture<BackendFixture>
         JsonNode listed = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/probes");
         Assert.Contains(listed["items"]!.AsArray(), p => p!["id"]!.GetValue<string>() == id && !p["revoked"]!.GetValue<bool>());
         string duplicateToken = await _backend.CreateJoinTokenAsync(role: "probe");
-        using var http = new HttpClient();
+        using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
         Fleet.Core.Result<Fleet.Enrollment.Client.EnrollmentResult> duplicate = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync(
             http, _backend.BaseAddress, duplicateToken, name, Path.Combine(Path.GetTempPath(), "moshiro-dup-" + Guid.NewGuid().ToString("N")), "0.0.0", null, TimeProvider.System, default);
         Assert.False(duplicate.IsSuccess);

backend/tests/MoShiro.Backend.Tests/HostTests.cs

@@ -184,6 +184,7 @@ public sealed class HostStatusTests : IClassFixture<BackendFixture>
     }
 
     [Fact]
+    [Trait("Category", "Slow")]
     public async Task ARealProbeReportsItsHostStatus_LikeAnAgent_NeverCarriesModelsOrAddresses_AndDisappearsWhenRevoked()
     {
         await _backend.SiteMustBePublicAsync();

backend/tests/MoShiro.Backend.Tests/NodeInfoTests.cs

@@ -174,6 +174,7 @@ public sealed partial class NodeInfoTests
 /// 访客能拿到的东西里不能有 IP 等敏感信息：公开接口的整个响应里不出现 IP 形状的字符串，也没有 ip、city、asn、经纬度这些字段。
 /// Agent 和 Probe 这里是真的连上了（Probe 的来源地址是 127.0.0.1），所以如果哪里把地址带出来了，这里一定能发现。
 /// </summary>
+[Trait("Category", "Slow")]
 public sealed partial class NodePrivacyTests : IClassFixture<BackendFixture>
 {
     private static readonly string[] s_forbiddenKeys = ["ip", "city", "asn", "latitude", "longitude", "cidr", "remote", "public_host", "geo_host", "host_name", "hostname"];

backend/tests/MoShiro.Backend.Tests/PasskeyTests.cs

@@ -169,7 +169,7 @@ public sealed class PasskeyTests : IClassFixture<BackendFixture>
         Assert.Equal("no-store", login.Headers.CacheControl?.ToString());
 
         // 拿这个会话令牌就是登录状态，能力和密码登录一样
-        using var signedIn = new HttpClient { BaseAddress = _backend.BaseAddress };
+        using var signedIn = new HttpClient { BaseAddress = _backend.BaseAddress, Timeout = TimeSpan.FromSeconds(15) };
         signedIn.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", body["session_token"]!.GetValue<string>());
         JsonNode me = await JsonOf(await signedIn.GetAsync("/api/admin/auth/me"));
         Assert.Equal(_user, me["username"]!.GetValue<string>());

backend/tests/MoShiro.Backend.Tests/ProbeHarness.cs

@@ -38,7 +38,7 @@ public sealed class ProbeHarness : IAsyncDisposable
 
         ProbeFile file = Fleet.Hosting.NodeConfig.Load(path, ProbeConfigContext.Default.ProbeFile);
         configure?.Invoke(file);
-        var app = new ProbeApp(file, enrolled.Identity, directory, TimeProvider.System, new Fleet.Hosting.ConsoleNodeLog(TimeProvider.System, Console.Out));
+        var app = new ProbeApp(file, enrolled.Identity, directory, TimeProvider.System, NullNodeLog.Instance);
         return new ProbeHarness(directory, app, enrolled) { ProbeId = enrolled.ProbeId };
     }
 

backend/tests/MoShiro.Backend.Tests/ReleaseFeedTests.cs

@@ -149,7 +149,7 @@ public sealed class ReleaseFeedTests : IAsyncLifetime
         await panel.StartAsync();
         try
         {
-            using var http = new HttpClient { BaseAddress = new Uri(panel.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()) };
+            using var http = new HttpClient { BaseAddress = new Uri(panel.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) };
             using HttpResponseMessage setup = await http.PostAsync("/api/admin/auth/setup", BackendFixture.Json(new JsonObject { ["claim_code"] = "FEED-TEST-CODE", ["username"] = "owner", ["password"] = "correct horse battery", ["return_session"] = true }));
             Assert.Equal(HttpStatusCode.Created, setup.StatusCode);
             http.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", JsonNode.Parse(await setup.Content.ReadAsStringAsync())!["session_token"]!.GetValue<string>());

backend/tests/MoShiro.Backend.Tests/SetupTests.cs

@@ -35,7 +35,7 @@ public sealed class SetupTests : IAsyncLifetime
                 options.GeoLookupUrl = string.Empty;
             });
         await _app.StartAsync();
-        _http = new HttpClient { BaseAddress = new Uri(_app.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()) };
+        _http = new HttpClient { BaseAddress = new Uri(_app.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) };
     }
 
     public async ValueTask DisposeAsync()
@@ -142,7 +142,7 @@ public sealed class SetupTests : IAsyncLifetime
         await generated.StartAsync();
         try
         {
-            using var http = new HttpClient { BaseAddress = new Uri(generated.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()) };
+            using var http = new HttpClient { BaseAddress = new Uri(generated.Services.GetRequiredService<IServer>().Features.Get<IServerAddressesFeature>()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) };
             foreach (string guess in new[] { Code, "0000-0000-0000-0000", "admin", "TEST-CLAIM-CODE-0001" })
             {
                 using HttpResponseMessage response = await http.PostAsync("/api/admin/auth/setup", BackendFixture.Json(new JsonObject { ["claim_code"] = guess, ["username"] = "owner", ["password"] = "correct horse battery" }));

backend/tests/MoShiro.Backend.Tests/TrustTests.cs

@@ -49,7 +49,7 @@ public sealed class TrustTests : IClassFixture<BackendFixture>
     private async Task<Result<EnrollmentResult>> EnrollAsync(string code, string? genesis, string[]? controllers, string? directory = null)
     {
         directory ??= Path.Combine(Path.GetTempPath(), "moshiro-trust-id-" + Guid.NewGuid().ToString("N"));
-        using var http = new HttpClient();
+        using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
         return await EnrollmentClient.EnrollAsync(http, _backend.BaseAddress, code, _backend.UniqueName("node"), directory, "0.0.0-test", null, TimeProvider.System, default, genesis, controllers);
     }
 
@@ -128,6 +128,7 @@ public sealed class TrustTests : IClassFixture<BackendFixture>
     }
 
     [Fact]
+    [Trait("Category", "Slow")]
     public async Task APinnedAgent_FollowsTheOperatorSet_WhenThePasskeysChange_AndOnlyWhenTheOperatorSigns()
     {
         Operator op = await OperatorAsync();
@@ -215,6 +216,7 @@ public sealed class TrustTests : IClassFixture<BackendFixture>
     }
 
     [Fact]
+    [Trait("Category", "Slow")]
     public async Task APinnedAgent_ReportsItsOwner_AndThePanelFindsTheAccountAndPasskey()
     {
         // 宿主（属主）要求在节点这一侧：节点把安装时固定的属主报上来，面板顺着凭证 id 反查出它绑定在哪个账户的哪把 passkey 上

backend/tests/MoShiro.TestSupport/ProbeSimulator.cs

@@ -70,8 +70,10 @@ public sealed class ProbeSimulator : IAsyncDisposable
 
     public async Task ConnectAsync(string host, int controlPort, byte[] agentPublicKey, int udpPort, NodeId agentId, TimeProvider? time = null, CancellationToken ct = default)
     {
+        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        timeout.CancelAfter(TimeSpan.FromSeconds(10));
         _agentId = agentId;
-        (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, ct);
+        (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, timeout.Token);
         _info = info;
         _agentKey = agentPublicKey;
         _dataSession = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
@@ -88,8 +90,8 @@ public sealed class ProbeSimulator : IAsyncDisposable
         _rpc.OnBinary = (_, _) => ValueTask.CompletedTask;
         _run = _rpc.RunAsync(_stop.Token);
         _receiver = _quic.RunAsync(_stop.Token);
-        await SyncClockAsync();
-        await _quic.ConnectAsync(ct);
+        await SyncClockAsync(ct: timeout.Token);
+        await _quic.ConnectAsync(timeout.Token);
     }
 
     public async Task<LineTaskSet> WaitForTasksAsync(Func<LineTaskSet, bool> predicate, TimeSpan? timeout = null)
@@ -111,16 +113,27 @@ public sealed class ProbeSimulator : IAsyncDisposable
     }
 
     /// <summary>NTP 式的时钟同步：偏移 = Agent 时钟 − (往返中点的) Probe 时钟。取最小往返的一次。</summary>
-    public async Task SyncClockAsync(int rounds = 5)
+    public async Task SyncClockAsync(int rounds = 3, CancellationToken ct = default)
     {
         long bestRoundTrip = long.MaxValue;
         for (int i = 0; i < rounds; i++)
         {
             long t0 = NowUs();
-            Result<ClockResult> reply = await _rpc!.RequestAsync(MoShiroMethods.Clock, new ClockParams { ProbeMonoUs = t0 }, MoShiroJsonContext.Default.ClockParams, MoShiroJsonContext.Default.ClockResult, CancellationToken.None);
+            Result<ClockResult> reply = await _rpc!.RequestAsync(
+                MoShiroMethods.Clock,
+                new ClockParams { ProbeMonoUs = t0 },
+                MoShiroJsonContext.Default.ClockParams,
+                MoShiroJsonContext.Default.ClockResult,
+                ct,
+                TimeSpan.FromSeconds(2));
             long t1 = NowUs();
+            if (!reply.IsSuccess)
+            {
+                continue;
+            }
+
             long roundTrip = t1 - t0;
-            if (reply.IsSuccess && roundTrip < bestRoundTrip)
+            if (roundTrip < bestRoundTrip)
             {
                 bestRoundTrip = roundTrip;
                 ClockOffsetUs = reply.Value.AgentMonoUs - ((t0 + t1) / 2);
@@ -164,6 +177,9 @@ public sealed class ProbeSimulator : IAsyncDisposable
     /// <summary>对一个 tcp_echo 线路任务建立一条 TCP（可选 TLS）连接，发 <paramref name="count"/> 条回显消息，然后交出一个列块。</summary>
     public async Task<(int Sent, int Received)> RunTcpEchoAsync(LineTask task, int tcpPort, int count, TimeSpan interval, bool tls, CancellationToken ct = default)
     {
+        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        timeout.CancelAfter(TimeSpan.FromSeconds(15));
+        ct = timeout.Token;
         using var client = new System.Net.Sockets.TcpClient();
         await client.ConnectAsync(IPAddress.Loopback, tcpPort, ct);
         Stream stream = client.GetStream();
@@ -246,10 +262,20 @@ public sealed class ProbeSimulator : IAsyncDisposable
     public async ValueTask DisposeAsync()
     {
         await _stop.CancelAsync();
-        _quic?.Dispose();
+        if (_quic is not null)
+        {
+            await _quic.DisposeAsync();
+        }
         if (_rpc is not null)
         {
-            await _rpc.DisposeAsync();
+            try
+            {
+                await _rpc.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2));
+            }
+            catch (Exception ex) when (ex is TimeoutException or OperationCanceledException or IOException or ObjectDisposedException)
+            {
+                // 测试连接关闭属于尽力清理，不能让一条卡住的流拖住整个测试程序集。
+            }
         }
 
         foreach (Task? task in new[] { _run, _receiver })