using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using System.Text.Json;
using Fleet.Core;
using Fleet.Identity;
using Fleet.Link;
using Fleet.Rpc;
using MoShiro.Measurement;
using MoShiro.Protocol;

namespace MoShiro.Probe;

/// <summary>所有连接共用的东西。</summary>
public sealed record ConnectionEnv(NodeIdentity Identity, string ProbeId, ProbeLimits Limits, AgentDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log);

/// <summary>
/// 与一个 Agent 的点对点会话：认证握手 → 同步时钟 → 按 Agent 下发的 <c>line_task_set</c> 运行角色 →
/// 每 10 秒把已完成的行切成列块放进发件箱 → 响应按需测试的阶段。会话断开时所有角色停止（凭据属于会话），按退避重连。
/// 会话内的所有状态（套接字、发件箱、角色）随会话一起创建和释放。
/// </summary>
public sealed class AgentConnection : IDisposable
{
    private static readonly TimeSpan s_flushInterval = TimeSpan.FromSeconds(ProtocolLimits.ChunkSeconds);
    private static readonly TimeSpan s_clockInterval = TimeSpan.FromSeconds(30);

    /// <summary>Agent 的任何帧（时钟回复、任务集）都算存活；比时钟同步间隔长得多，容得下差线路上的黑洞，但不会让一个 Agent 已经丢掉的会话永远挂着。</summary>
    private static readonly TimeSpan s_idleLimit = TimeSpan.FromSeconds(120);

    /// <summary>一直在发测量报文、却这么久没有收到 Agent 的任何一个：数据面已经不通，重建会话。要比一次线路中断长（中断恢复后接着测），又不能拖到几个小时。</summary>
    private static readonly TimeSpan s_dataIdleLimit = TimeSpan.FromSeconds(90);
    private static readonly RpcRouter<ConnectionSession> s_router = BuildRouter();

    /// <summary>对等面入站方法的路由器（只读，测试用）。</summary>
    public static RpcRouter<ConnectionSession> Router => s_router;

    private readonly ConnectionEnv _env;
    private volatile ConnectionSession? _session;
    private int _previousQuicSourcePort;
    private QuicDatagramTransport? _activeDatagrams;
    private readonly object _handshakeGate = new();
    private readonly Queue<double> _handshakeSamplesMs = new();
    private readonly SemaphoreSlim _retrySignal = new(0, 1);
    private ProbeLinkAttemptState _attemptState = ProbeLinkAttemptState.Connecting;
    private string? _lastError;
    private DateTimeOffset? _retryAt;
    private long _handshakeAttempts;
    private long _handshakeSuccesses;
    private long _handshakeFailures;
    private long _handshakeTimeouts;

    public AgentConnection(KnownAgent endpoint, ConnectionEnv env)
    {
        Endpoint = endpoint;
        _env = env;
    }

    public KnownAgent Endpoint { get; }

    public bool IsConnected => _session is not null && Volatile.Read(ref _activeDatagrams)?.IsConnected == true;

    public int RoleCount => _session?.RoleCount ?? 0;

    public long ClockOffsetUs => _session?.Clock.OffsetUs ?? 0;

    /// <summary>管理端手动配对时，跳过当前退避等待，马上再拨一次。</summary>
    public void RetryNow()
    {
        if (IsConnected)
        {
            return;
        }
        try
        {
            _retrySignal.Release();
        }
        catch (SemaphoreFullException)
        {
            // 已经有一次手动重试排队。
        }
        catch (ObjectDisposedException)
        {
            // 对应线路正在被移除。
        }
    }

    public void Dispose() => _retrySignal.Dispose();

    public ProbeQuicLinkHealth QuicHealth()
    {
        lock (_handshakeGate)
        {
            double? p95 = null;
            if (_handshakeSamplesMs.Count > 0)
            {
                double[] sorted = [.. _handshakeSamplesMs.Order()];
                p95 = sorted[(int)Math.Ceiling(sorted.Length * 0.95) - 1];
            }

            QuicDatagramTransport? datagrams = Volatile.Read(ref _activeDatagrams);
            bool connected = _session is not null && datagrams?.IsConnected == true;
            ProbeLinkAttemptState state = _attemptState;
            string? lastError = _lastError;
            if (!connected && state == ProbeLinkAttemptState.Connected)
            {
                state = ProbeLinkAttemptState.RetryWait;
                lastError ??= "QUIC datagram transport is closed; reconnecting";
            }

            QuicTransportSnapshot? transport = datagrams?.GetTransportSnapshot();
            return new ProbeQuicLinkHealth
            {
                AgentId = Endpoint.AgentId,
                IpVersion = Endpoint.IpVersion,
                Connected = connected,
                State = state,
                LastError = lastError,
                RetryAttempt = _lastRetryAttempt,
                RetryInMs = state == ProbeLinkAttemptState.RetryWait && _retryAt is { } retryAt
                    ? Math.Max(0, (long)(retryAt - _env.Time.GetUtcNow()).TotalMilliseconds)
                    : null,
                HandshakeAttempts = _handshakeAttempts,
                HandshakeSuccesses = _handshakeSuccesses,
                HandshakeFailures = _handshakeFailures,
                HandshakeTimeouts = _handshakeTimeouts,
                HandshakeP95Ms = p95,
                HandshakeSampleCount = _handshakeSamplesMs.Count,
                TransportPacketLossRate = transport?.PacketLossRate,
                TransportPacketsSent = transport?.PacketsSent,
                TransportCongestionEvents = transport?.CongestionEvents,
                TransportEcnCongestionEvents = transport?.EcnCongestionEvents,
                PathMtu = transport?.PathMtu,
            };
        }
    }

    private int _lastRetryAttempt;

    public async Task RunAsync(CancellationToken ct)
    {
        var backoff = new ProbeReconnectBackoff();
        while (!ct.IsCancellationRequested)
        {
            SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0);
            try
            {
                string reason = await RunOnceAsync(backoff, ct);
                SetAttempt(ProbeLinkAttemptState.RetryWait, reason, null, backoff.Attempts);
                _env.Log.Warn("agent session ended", ("agent", Endpoint.AgentId), ("err", reason));
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return;
            }
            catch (Exception ex)
            {
                SetAttempt(ProbeLinkAttemptState.RetryWait, $"{ex.GetType().Name}: {ex.Message}", null, backoff.Attempts + 1);
                _env.Log.Error("agent connection attempt failed; retry will continue", ("agent", Endpoint.AgentId), ("exception", ex.GetType().FullName), ("err", ex.Message));
            }

            try
            {
                TimeSpan delay = backoff.Next();
                lock (_handshakeGate)
                {
                    _attemptState = ProbeLinkAttemptState.RetryWait;
                    _lastRetryAttempt = backoff.Attempts;
                    _retryAt = _env.Time.GetUtcNow() + delay;
                }
                using var waitCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
                Task delayTask = Task.Delay(delay, _env.Time, waitCts.Token);
                Task retryTask = _retrySignal.WaitAsync(waitCts.Token);
                Task completed = await Task.WhenAny(delayTask, retryTask);
                await waitCts.CancelAsync();
                if (completed == retryTask)
                {
                    await retryTask;
                    _env.Log.Info("manual link retry requested", ("agent", Endpoint.AgentId));
                }
                else
                {
                    await delayTask;
                    try { await retryTask; }
                    catch (OperationCanceledException) { }
                }
            }
            catch (OperationCanceledException)
            {
                return;
            }
        }
    }

    private async Task<string> RunOnceAsync(ProbeReconnectBackoff backoff, CancellationToken ct)
    {
        AddressFamily family = Endpoint.IpVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork;
        using var resolveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), _env.Time);
        using CancellationTokenSource resolveCt = CancellationTokenSource.CreateLinkedTokenSource(ct, resolveTimeout.Token);
        IPAddress[] addresses = IPAddress.TryParse(Endpoint.Host, out IPAddress? literal)
            ? [literal]
            : await Dns.GetHostAddressesAsync(Endpoint.Host, family, resolveCt.Token);
        IPAddress? address = addresses.FirstOrDefault(a => a.AddressFamily == family && !a.IsIPv4MappedToIPv6);
        if (address is null)
        {
            throw new SocketException((int)SocketError.HostNotFound);
        }

        byte[] agentKey = Convert.FromBase64String(Endpoint.PublicKey);
        (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(
            address.ToString(), Endpoint.TcpPort, agentKey, _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct);
        var data = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
        var remote = (IPEndPoint)info.Remote!;
        IReadOnlyList<int> quicPorts = Endpoint.EffectiveQuicPorts;
        int quicPort = quicPorts.Count == 0 ? 7443 : quicPorts[0];
        var datagrams = new QuicDatagramTransport(
            data,
            _env.Clock,
            new DataPlaneCounters(),
            _env.Log,
            new IPEndPoint(remote.Address, quicPort),
            agentKey,
            Volatile.Read(ref _previousQuicSourcePort));
        Volatile.Write(ref _activeDatagrams, datagrams);
        datagrams.Bind();
        using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(ct);

        var outbox = new Outbox(8 << 20);
        var clock = new ClockEstimator();
        var shipper = new FragmentShipper(outbox, new NodeId(Endpoint.AgentId), new ProbeId(_env.ProbeId), clock, _env.Time);
        var blockEnv = new BlockEnv
        {
            Clock = _env.Clock,
            Log = _env.Log,
            Session = () => data,
            Datagrams = datagrams,
            PeerAddress = remote.Address,
            PeerSpkiSha256 = NodeIdentity.Fingerprint(agentKey),
            KnockKey = Knock.DeriveKey(_env.Identity, agentKey),
            Fragments = shipper,
            MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
            MaxBulkMbps = _env.Limits.MaxTestMbps,
        };
        var session = new ConnectionSession(blockEnv, clock, outbox, _env);
        var rpc = new RpcSession<ConnectionSession>(transport, s_router, session, _env.Time);
        rpc.OnBinary = (_, _) => ValueTask.CompletedTask;

        Task run = rpc.RunAsync(cts.Token);
        Task receive = datagrams.RunAsync(cts.Token);
        Task pump = PumpAsync(rpc, outbox, cts.Token);
        try
        {
            // 先同步时钟（块头里要带偏移）；Agent 随后推送线路任务集
            await SyncClockAsync(rpc, clock, rounds: 5, cts.Token);
            lock (_handshakeGate)
            {
                _handshakeAttempts++;
            }

            try
            {
                SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0);
                await datagrams.ConnectAsync(cts.Token);
                lock (_handshakeGate)
                {
                    _handshakeSuccesses++;
                    if (datagrams.HandshakeDurationMs is { } elapsed)
                    {
                        _handshakeSamplesMs.Enqueue(elapsed);
                        while (_handshakeSamplesMs.Count > 128)
                        {
                            _handshakeSamplesMs.Dequeue();
                        }
                    }
                }
            }
            catch (Exception ex) when (!(ex is OperationCanceledException && cts.IsCancellationRequested))
            {
                lock (_handshakeGate)
                {
                    _handshakeFailures++;
                    if (ex is TimeoutException)
                    {
                        _handshakeTimeouts++;
                    }
                }

                throw;
            }
            if (datagrams.BoundPorts.Count > 0)
            {
                Volatile.Write(ref _previousQuicSourcePort, datagrams.BoundPorts[0]);
            }

            backoff.Reset();
            _retrySignal.Wait(0, CancellationToken.None); // 若“立即重试”恰好赶在本次拨号成功前到达，本次成功已满足请求。
            _session = session;
            SetAttempt(ProbeLinkAttemptState.Connected, null, null, 0);
            _env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs));
            Task maintenance = MaintainAsync(rpc, session, clock, datagrams, cts.Token);
            Task finished = await Task.WhenAny(run, receive, pump, maintenance);
            string reason = finished == run
                ? "closed by the Agent control connection"
                : finished == receive
                    ? "QUIC datagram transport closed"
                    : finished == pump
                        ? "Probe outbox pump stopped"
                        : "connection maintenance stopped";
            if (finished.IsFaulted && finished.Exception?.GetBaseException() is { } error)
            {
                reason = $"{error.GetType().Name}: {error.Message}";
            }

            await cts.CancelAsync();
            await BoundedAsync("stop tasks", () => Task.WhenAll(Quiet(run), Quiet(receive), Quiet(pump), Quiet(maintenance)));
            return reason;
        }
        finally
        {
            _session = null;
            Volatile.Write(ref _activeDatagrams, null);
            await cts.CancelAsync();
            await BoundedAsync("stop blocks", session.StopAllAsync);
            await BoundedAsync("wait control", () => Quiet(run));
            await BoundedAsync("wait datagrams", () => Quiet(receive));
            await BoundedAsync("wait pump", () => Quiet(pump));
            await BoundedAsync("dispose control", () => rpc.DisposeAsync().AsTask());
            await BoundedAsync("dispose quic", () => datagrams.DisposeAsync().AsTask());
        }
    }

    private void SetAttempt(ProbeLinkAttemptState state, string? error, DateTimeOffset? retryAt, int attempt)
    {
        lock (_handshakeGate)
        {
            _attemptState = state;
            if (error is not null || state == ProbeLinkAttemptState.Connected)
            {
                _lastError = error;
            }
            _retryAt = retryAt;
            _lastRetryAttempt = attempt;
        }
    }

    private async Task BoundedAsync(string step, Func<Task> work)
    {
        if (!await Bounded.WaitAsync(work(), Bounded.Teardown, _env.Time))
        {
            _env.Log.Warn("agent session teardown step timed out, moving on", ("agent", Endpoint.AgentId), ("step", step));
        }
    }

    private static async Task Quiet(Task task)
    {
        try
        {
            await task;
        }
        catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or System.Net.Sockets.SocketException or InvalidOperationException)
        {
            // 会话结束时的正常噪声
        }
    }

    private static async Task PumpAsync(RpcSession<ConnectionSession> rpc, Outbox outbox, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            byte[] frame = await outbox.TakeNextAsync(2 << 20, ct);
            await rpc.SendBinaryAsync(frame, ct);
        }
    }

    private async Task SyncClockAsync(RpcSession<ConnectionSession> rpc, ClockEstimator clock, int rounds, CancellationToken ct)
    {
        for (int i = 0; i < rounds; i++)
        {
            long sent = _env.Clock.NowUs();
            Result<ClockResult> reply = await rpc.RequestAsync(MoShiroMethods.Clock, new ClockParams { ProbeMonoUs = sent }, MoShiroJsonContext.Default.ClockParams, MoShiroJsonContext.Default.ClockResult, ct);
            if (reply.TryGet(out ClockResult? value, out _))
            {
                clock.Add(sent, value.AgentMonoUs, _env.Clock.NowUs());
            }
        }
    }

    /// <summary>切块（每 10 秒）和时钟同步（每 30 秒）。</summary>
    private async Task MaintainAsync(RpcSession<ConnectionSession> rpc, ConnectionSession session, ClockEstimator clock, QuicDatagramTransport datagrams, CancellationToken ct)
    {
        using var flush = new PeriodicTimer(s_flushInterval, _env.Time);
        long lastClockUs = _env.Clock.NowUs();
        long seenIn = datagrams.Counters.PacketsIn;
        long seenOut = datagrams.Counters.PacketsOut;
        long inChangedUs = lastClockUs;
        while (await flush.WaitForNextTickAsync(ct))
        {
            if (rpc.IdleFor > s_idleLimit)
            {
                // Agent 已经不认这个会话（它重启、重连、或者空闲超时关了会话，而 FIN 没有到达）：Probe 这边结束它，让外层按退避重拨
                _env.Log.Warn("agent session is silent, reconnecting", ("agent", Endpoint.AgentId), ("idle_s", (long)rpc.IdleFor.TotalSeconds));
                throw new IOException("the Agent stopped answering on this session");
            }

            // 数据面也要有回应：控制连接（TCP）还活着，QUIC 这条路却可能早已黑洞（一直在发、什么也收不到），表现为去程 100% 丢包、没有回程
            long packetsIn = datagrams.Counters.PacketsIn;
            long packetsOut = datagrams.Counters.PacketsOut;
            if (packetsIn != seenIn)
            {
                seenIn = packetsIn;
                inChangedUs = _env.Clock.NowUs();
            }
            else if (packetsOut != seenOut && _env.Clock.NowUs() - inChangedUs > s_dataIdleLimit.TotalMicroseconds)
            {
                _env.Log.Warn("agent datagram path is silent, reconnecting", ("agent", Endpoint.AgentId), ("idle_s", (_env.Clock.NowUs() - inChangedUs) / 1_000_000));
                throw new IOException("the Agent has not answered any measurement datagram");
            }

            seenOut = packetsOut;
            session.FlushLineBlocks();
            if (_env.Clock.NowUs() - lastClockUs >= s_clockInterval.TotalMicroseconds)
            {
                lastClockUs = _env.Clock.NowUs();
                await SyncClockAsync(rpc, clock, rounds: 1, ct);
            }
        }
    }

    private static RpcRouter<ConnectionSession> BuildRouter() =>
        new RpcRouter<ConnectionSession>(RpcSide.Dialer, MoShiroMethodTable.Table)
            .On(
                FleetMethods.StateSync,
                FleetRpcContext.Default.StateSyncParams,
                FleetRpcContext.Default.StateSyncResult,
                async (session, sync, _) =>
                {
                    var result = new StateSyncResult { AppliedVersion = sync.Version };
                    if (sync.Type == LineTaskSet.StateType)
                    {
                        LineTaskSet? set = sync.Doc.Deserialize(MoShiroJsonContext.Default.LineTaskSet);
                        if (set is null)
                        {
                            return Result<StateSyncResult>.Fail(FleetErrors.InvalidParams, "line_task_set could not be parsed");
                        }

                        result.Rejected.AddRange(await session.ApplyAsync(set));
                    }

                    return Result<StateSyncResult>.Ok(result);
                })
            .On(
                MoShiroMethods.TestPhase,
                MoShiroJsonContext.Default.TestPhaseStartParams,
                MoShiroJsonContext.Default.TestPhaseStartResult,
                (session, start, _) => ValueTask.FromResult(Result<TestPhaseStartResult>.Ok(new TestPhaseStartResult { Accepted = session.StartPhase(start) })))
            .On(
                MoShiroMethods.TestPhaseEnd,
                MoShiroJsonContext.Default.TestPhaseEndParams,
                MoShiroJsonContext.Default.TestPhaseEndResult,
                async (session, end, ct) => Result<TestPhaseEndResult>.Ok(new TestPhaseEndResult { Fragments = await session.EndPhaseAsync(end, ct) }))
            .OnNotify(
                MoShiroMethods.TestAbort,
                MoShiroJsonContext.Default.TestAbortParams,
                async (session, abort, _) => await session.AbortAsync(abort.TestId))
            .OnNotify(
                FleetMethods.OutboxAck,
                FleetRpcContext.Default.OutboxAckParams,
                (session, ack, _) =>
                {
                    session.Outbox.Acknowledge(ack.Ids.Concat(ack.Duplicates).Select(Guid.Parse));
                    return ValueTask.CompletedTask;
                });
}

/// <summary>一次会话内的运行状态：线路任务的积木、按需测试的阶段。方法由路由器在单个入站队列里顺序调用。</summary>
public sealed class ConnectionSession
{
    private const int MaxBlocks = 16;

    private readonly object _gate = new();
    private readonly Dictionary<string, RunningBlock> _lineBlocks = [];
    private readonly ConcurrentDictionary<uint, PhaseState> _phases = new();

    public ConnectionSession(BlockEnv blockEnv, ClockEstimator clock, Outbox outbox, ConnectionEnv env)
    {
        BlockEnv = blockEnv;
        Clock = clock;
        Outbox = outbox;
        Env = env;
    }

    public BlockEnv BlockEnv { get; }

    public ClockEstimator Clock { get; }

    public Outbox Outbox { get; }

    public ConnectionEnv Env { get; }

    public int RoleCount
    {
        get
        {
            lock (_gate)
            {
                return _lineBlocks.Count;
            }
        }
    }

    /// <summary>调和线路任务：参数没有变化的积木继续运行，其余启动或停止。返回被拒绝的任务（不认识的 Kind、参数非法、超过上限）。</summary>
    public async Task<List<RejectedItem>> ApplyAsync(LineTaskSet set)
    {
        var rejected = new List<RejectedItem>();
        var wanted = new Dictionary<string, BlockSpec>();
        foreach (LineTask task in set.Tasks)
        {
            BlockSpec? spec = BlockCatalog.ProbeKinds.Contains(task.Kind)
                ? BlockSpec.Create(task.Kind, task.Params, task.StreamId, task.Port, new FragmentScope(task.Id, null, null))
                : null;
            if (spec is null)
            {
                rejected.Add(new RejectedItem { Id = task.Id.Value, Reason = BlockCatalog.ProbeKinds.Contains(task.Kind) ? "invalid_params" : "unsupported" });
            }
            else if (wanted.Count >= MaxBlocks)
            {
                rejected.Add(new RejectedItem { Id = task.Id.Value, Reason = "budget_exceeded" });
            }
            else
            {
                wanted[task.Id.Value] = spec;
            }
        }

        var stopped = new List<RunningBlock>();
        lock (_gate)
        {
            foreach ((string id, RunningBlock running) in _lineBlocks.ToList())
            {
                if (!wanted.TryGetValue(id, out BlockSpec? spec) || spec.Signature != running.Signature)
                {
                    stopped.Add(running);
                    _lineBlocks.Remove(id);
                }
            }

            foreach ((string id, BlockSpec spec) in wanted)
            {
                if (!_lineBlocks.ContainsKey(id))
                {
                    _lineBlocks[id] = RunningBlock.Start(BlockCatalog.Find(spec.KindId)!.Dialer(BlockEnv, spec), spec.Signature);
                }
            }
        }

        // 被停止的积木把已经稳定的行交出去（刚发出、还在等回复的行不交：它们不是丢失，只是没有等到）
        foreach (RunningBlock running in stopped)
        {
            await running.StopAsync();
            running.Block.Flush(FlushMode.Settled);
        }

        return rejected;
    }

    public void FlushLineBlocks()
    {
        List<RunningBlock> blocks;
        lock (_gate)
        {
            blocks = [.. _lineBlocks.Values];
        }

        foreach (RunningBlock running in blocks)
        {
            running.Block.Flush(FlushMode.Settled);
        }
    }

    /// <summary>按需测试的一个阶段：起对应的拨号方积木（单向数据流是接收端）。超过本地上限（或不认识）则拒绝。</summary>
    public bool StartPhase(TestPhaseStartParams start)
    {
        string? direction = start.Params.TryGetValue("direction", out System.Text.Json.JsonElement d) ? d.GetString() : null;
        if (_phases.ContainsKey(start.StreamId) || !BlockCatalog.ProbeKinds.Contains(start.Kind) || BlockCatalog.Find(start.Kind)!.Supports(direction) is false)
        {
            return false;
        }

        if (start.Kind == "udp_train")
        {
            double mbps = start.Params["mbps"].GetDouble();
            long size = start.Params["size"].GetInt64();
            long pps = (long)(mbps * 1_000_000 / 8 / Math.Max(1, size));
            if (mbps > Env.Limits.MaxTestMbps || pps > Env.Limits.MaxTotalPps)
            {
                return false;
            }
        }

        BlockSpec? spec = BlockSpec.Create(start.Kind, start.Params, start.StreamId, start.Port, new FragmentScope(null, start.TestId, start.Phase));
        if (spec is null)
        {
            return false;
        }

        _phases[start.StreamId] = new PhaseState(start.TestId, start.Kind, RunningBlock.Start(BlockCatalog.Find(start.Kind)!.Dialer(BlockEnv, spec), spec.Signature));
        return true;
    }

    /// <summary>阶段结束：收尾、交出列块，返回块数。</summary>
    public async Task<int> EndPhaseAsync(TestPhaseEndParams end, CancellationToken ct)
    {
        if (!_phases.TryRemove(end.StreamId, out PhaseState? phase))
        {
            return 0;
        }

        // 最后一批包（单向数据流）、最后几个回显的回复还在途中
        await Task.Delay(TimeSpan.FromMilliseconds(phase.Kind == "udp_train" ? 200 : 300), BlockEnv.Time, ct);
        await phase.Running.StopAsync();
        return phase.Running.Block.Flush(FlushMode.All, end.SentCount);
    }

    public async Task AbortAsync(TestId test)
    {
        foreach ((uint stream, PhaseState phase) in _phases.ToList())
        {
            if (phase.Test == test && _phases.TryRemove(stream, out _))
            {
                await phase.Running.StopAsync();
            }
        }
    }

    public async Task StopAllAsync()
    {
        List<RunningBlock> blocks;
        lock (_gate)
        {
            blocks = [.. _lineBlocks.Values];
            _lineBlocks.Clear();
        }

        blocks.AddRange(_phases.Values.Select(p => p.Running));
        _phases.Clear();
        foreach (RunningBlock running in blocks)
        {
            await running.StopAsync();
        }
    }

    private sealed record PhaseState(TestId Test, string Kind, RunningBlock Running);
}