fleet: require a minimum version of the peer at the handshake [CI]

MoShiro__MinAgentVersion makes the backend refuse agents below it at hello (fleet_version_required); fleet.min_server_version or MOSHIRO_MIN_SERVER_VERSION makes the agent drop a backend below it. Either side waits five seconds and shakes hands again, because the other one may be updating.

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

@@ -28,7 +28,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
     private readonly LoadedNodeIdentity _identity;
     private readonly TimeProvider _time;
     private readonly INodeLog _log;
-    private readonly INodeLogBatchSource? _logBatchSource;
+    private readonly INodeLogSource? _logSource;
     private readonly MonoClock _clock;
     private readonly AgentCounters _counters = new();
     private readonly SessionRegistry _sessions = new();
@@ -62,7 +62,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
         _identity = identity;
         _time = time;
         _log = log;
-        _logBatchSource = log as INodeLogBatchSource;
+        _logSource = log as INodeLogSource;
         _clock = new MonoClock(time);
         _uplink = new BackendUplink(time);
 
@@ -320,6 +320,23 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
                 async (_, request, _) => Result<TestRunResult>.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.DialerId.Value) }));
         }
 
+        if (Allow(MoShiroMethods.LogsFetch))
+        {
+            router.On(
+                MoShiroMethods.LogsFetch,
+                MoShiroJsonContext.Default.LogsFetchParams,
+                MoShiroJsonContext.Default.LogsFetchResult,
+                (_, request, _) =>
+                {
+                    NodeLogPage page = _logSource?.Read(request.AfterId, request.Limit, request.MinLevel) ?? new NodeLogPage([], 0);
+                    return ValueTask.FromResult(Result<LogsFetchResult>.Ok(new LogsFetchResult
+                    {
+                        LastId = page.LastId,
+                        Items = [.. page.Items.Select(e => new LogEntryDto { Id = e.Id, Type = e.Type, AtUtcMs = e.Record.AtUtcMs, Level = e.Record.Level, Message = e.Record.Message, Fields = e.Record.Fields })],
+                    }));
+                });
+        }
+
         if (Allow(MoShiroMethods.LinkRetry))
         {
             router.OnNotify(MoShiroMethods.LinkRetry, MoShiroJsonContext.Default.LinkRetryParams, (_, request, _) =>
@@ -336,6 +353,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
             ServerVerifier = _trust,
             NodeId = _identity.NodeId,
             NodeVersion = Version,
+            MinServerVersion = () => Environment.GetEnvironmentVariable("MOSHIRO_MIN_SERVER_VERSION") is { Length: > 0 } required ? required : _config.Fleet.MinServerVersion,
             Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', _listening ? BlockCatalog.AgentKinds : BlockCatalog.TargetKinds)],
             HelloMeta = HelloMeta,
             HeartbeatExtra = HeartbeatExtra,
@@ -407,7 +425,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
         long seconds = Math.Max(1, (nowMs - _lastAtMs) / 1000);
         long packetsIn = _counters.PacketsIn;
         long packetsOut = _counters.PacketsOut;
-        NodeLogBatch logBatch = _logBatchSource?.TakeBatch(16) ?? new NodeLogBatch([], 0);
+        NodeLogSummary? logSummary = _logSource?.TakeSummary();
         UpdateStatus? updateStatus = _nativeUpdate?.Status() ?? _update?.Status();
         var extra = new NodeHeartbeatExtra
         {
@@ -421,8 +439,14 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
             DroppedRateLimited = _counters.DroppedRateLimited,
             QueueBytes = _outbox.PendingBytes,
             Update = updateStatus,
-            RuntimeLogs = logBatch.Records.Count == 0 ? null : [.. logBatch.Records],
-            DroppedRuntimeLogs = logBatch.Dropped == 0 ? null : logBatch.Dropped,
+            LogSummary = logSummary is null ? null : new LogSummaryDto
+            {
+                LastId = logSummary.LastId,
+                Info = logSummary.Info,
+                Warn = logSummary.Warn,
+                Error = logSummary.Error,
+                Types = [.. logSummary.Types.Select(t => new LogTypeCountDto { Type = t.Type, Level = t.Level, Count = t.Count })],
+            },
         };
         _lastIn = packetsIn;
         _lastOut = packetsOut;

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

@@ -40,6 +40,10 @@ public sealed class FleetSection
     /// <summary>节点自己要求有属主（passkey）：没有就不运行。见 <c>OwnerGuard</c>。</summary>
     [JsonPropertyName("require_owner")]
     public bool RequireOwner { get; set; } = true;
+
+    /// <summary>面板至少要是哪个版本才继续连接（如 <c>0.4.2-dev.80</c>）；空 = 不要求。不够就不发任何东西，5 秒后重新握手（面板可能正在更新）。环境变量 <c>MOSHIRO_MIN_SERVER_VERSION</c> 优先。</summary>
+    [JsonPropertyName("min_server_version")]
+    public string? MinServerVersion { get; set; }
 }
 
 public sealed class ListenSection

backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs

@@ -217,7 +217,7 @@ public sealed class StreamInfo
         KindSpec spec = Kinds.Find(kind)!;
         DeclaredPps = kind switch
         {
-            "udp_echo" or "tcp_echo" => parameters.GetInt("rate_hz"),
+            "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kind, parameters),
             "udp_train" => parameters.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, parameters.GetInt("size")),
             _ => 20,
         };

backend/agent/src/MoShiro.Agent/Dial/ControlBackoff.cs

@@ -17,6 +17,8 @@ public sealed class ControlBackoff
         TimeSpan.FromSeconds(32),
     ];
 
+    private const int MaxMinutes = 10;
+
     private int _attempt;
 
     public int Attempts => Volatile.Read(ref _attempt);
@@ -24,7 +26,13 @@ public sealed class ControlBackoff
     public TimeSpan Next()
     {
         int attempt = Interlocked.Increment(ref _attempt) - 1;
-        return attempt < s_delays.Length ? s_delays[attempt] : TimeSpan.FromMinutes(1);
+        if (attempt < s_delays.Length)
+        {
+            return s_delays[attempt];
+        }
+
+        // 表走完还连不上（对端关了、这台机器没有这条路由）：一分钟起继续翻倍，最长十分钟，不再每分钟白白重试一次、写三条日志
+        return TimeSpan.FromMinutes(Math.Min(MaxMinutes, 1 << Math.Min(attempt - s_delays.Length, 4)));
     }
 
     public void Reset() => Interlocked.Exchange(ref _attempt, 0);

backend/agent/src/MoShiro.Agent/Dial/QuicRedialBackoff.cs

@@ -8,6 +8,8 @@ public sealed class QuicRedialBackoff
 {
     private static readonly TimeSpan[] s_delays = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(3), TimeSpan.FromSeconds(5)];
 
+    private const int ShortAttempts = 12;
+
     private int _attempt;
 
     public int Attempts => _attempt;
@@ -15,6 +17,12 @@ public sealed class QuicRedialBackoff
     public TimeSpan Next(Random? random = null)
     {
         TimeSpan delay = s_delays[Math.Min(_attempt, s_delays.Length - 1)];
+        if (_attempt >= ShortAttempts)
+        {
+            // 短间隔已经连续失败了这么多次：这条路径大概率是死的（对端不通、被丢弃），握手和日志都是白费，翻倍退避，最长五分钟
+            delay = TimeSpan.FromSeconds(Math.Min(300, 5 * (1 << Math.Min(_attempt - ShortAttempts + 1, 6))));
+        }
+
         _attempt++;
         return random is null ? delay : delay * (0.8 + (0.4 * random.NextDouble()));
     }

backend/agent/tests/MoShiro.Agent.Tests/ControlBackoffTests.cs

@@ -4,7 +4,7 @@ namespace MoShiro.Agent.Tests;
 public sealed class ControlBackoffTests
 {
     [Fact]
-    public void Next_UsesRequestedEscalationThenCapsAtOneMinute()
+    public void Next_UsesRequestedEscalationThenKeepsDoublingToTenMinutes()
     {
         var backoff = new ControlBackoff();
         TimeSpan[] expected =
@@ -13,7 +13,8 @@ public sealed class ControlBackoffTests
             TimeSpan.FromSeconds(8), TimeSpan.FromSeconds(16),
             TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32),
             TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32),
-            TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(1),
+            TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(4),
+            TimeSpan.FromMinutes(8), TimeSpan.FromMinutes(10), TimeSpan.FromMinutes(10),
         ];
 
         Assert.Equal(expected, expected.Select(_ => backoff.Next()));

backend/agent/tests/MoShiro.Agent.Tests/LinkStateTests.cs

@@ -16,6 +16,10 @@ public sealed class LinkStateTests
         int[] seconds = [1, 2, 3, 5, 5, 5];
         Assert.Equal(seconds, seconds.Select(_ => (int)backoff.Next().TotalSeconds));
 
+        // 连续失败很多次（路径是死的）：翻倍退避，最长五分钟
+        int[] later = [.. Enumerable.Range(0, 12).Select(_ => (int)backoff.Next().TotalSeconds)];
+        Assert.Equal([5, 5, 5, 10, 20, 40, 80, 160, 300], later[3..]);
+
         backoff.Reset();
 
         Assert.Equal(TimeSpan.FromSeconds(1), backoff.Next());

backend/fleet/src/Fleet.Core/ErrorCode.cs

@@ -27,4 +27,7 @@ public static class FleetErrors
     public static readonly ErrorCode JoinTokenExpired = new("fleet_join_token_expired");
     public static readonly ErrorCode ProofInvalid = new("fleet_proof_invalid");
     public static readonly ErrorCode FrameTooLarge = new("fleet_frame_too_large");
+
+    /// <summary>对方的版本低于要求的最低版本：等一会儿（对方可能正在更新）再握手。</summary>
+    public static readonly ErrorCode VersionRequired = new("fleet_version_required");
 }

backend/fleet/src/Fleet.Core/NodeSupport.cs

@@ -15,15 +15,27 @@ public interface INodeLog
     void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields);
 }
 
-/// <summary>One structured node log line queued for delivery to the control plane.</summary>
+/// <summary>One structured node log line.</summary>
 public sealed record NodeLogRecord(long AtUtcMs, string Level, string Message, Dictionary<string, string> Fields);
 
-public sealed record NodeLogBatch(IReadOnlyList<NodeLogRecord> Records, long Dropped);
+/// <summary>A retained log line: <paramref name="Id"/> grows with every line, <paramref name="Type"/> is a stable hash of the message text (the template), so counters can be sent without the text.</summary>
+public sealed record NodeLogEntry(long Id, uint Type, NodeLogRecord Record);
 
-/// <summary>Optional bounded log source used by node heartbeats to upload recent local logs.</summary>
-public interface INodeLogBatchSource
+/// <summary>How many lines of one type and level were written since the last summary.</summary>
+public sealed record NodeLogTypeCount(uint Type, string Level, long Count);
+
+/// <summary>What a node tells the control plane about its logs without sending any text: counters since the last summary and the newest line id.</summary>
+public sealed record NodeLogSummary(long LastId, long Info, long Warn, long Error, IReadOnlyList<NodeLogTypeCount> Types);
+
+public sealed record NodeLogPage(IReadOnlyList<NodeLogEntry> Items, long LastId);
+
+/// <summary>Bounded local log retention. The control plane gets counters in heartbeats and pulls the text when it needs it.</summary>
+public interface INodeLogSource
 {
-    NodeLogBatch TakeBatch(int maxRecords);
+    NodeLogSummary TakeSummary();
+
+    /// <summary>Lines with an id above <paramref name="afterId"/>, oldest first, at most <paramref name="limit"/>, at or above <paramref name="minLevel"/> (info, warn or error).</summary>
+    NodeLogPage Read(long afterId, int limit, string? minLevel);
 }
 
 public sealed class NullNodeLog : INodeLog

backend/fleet/src/Fleet.Core/NodeVersions.cs

@@ -0,0 +1,87 @@
+using System.Globalization;
+
+namespace Fleet.Core;
+
+/// <summary>
+/// 版本号比较，给「要求的最低版本」用。认得 <c>主.次.修订</c>，可以带预发布后缀；开发构建 <c>0.4.2-dev.73-5245f03</c> 按 <c>dev.73</c> 的序号比较，
+/// 提交哈希不参与比较。同一个主次修订里，正式版高于任何预发布版。认不出来的版本号不拒绝（返回 false），免得一个奇怪的版本字符串把节点挡在门外。
+/// </summary>
+public static class NodeVersions
+{
+    private readonly record struct Parsed(int Major, int Minor, int Patch, bool Prerelease, long Dev, string Suffix);
+
+    /// <summary><paramref name="actual"/> 低于 <paramref name="required"/> 返回 true；任何一个为空或无法解析返回 false。</summary>
+    public static bool IsBelow(string? actual, string? required)
+    {
+        if (string.IsNullOrWhiteSpace(required) || string.IsNullOrWhiteSpace(actual))
+        {
+            return false;
+        }
+
+        if (!TryParse(actual, out Parsed have) || !TryParse(required, out Parsed need))
+        {
+            return false;
+        }
+
+        return Compare(have, need) < 0;
+    }
+
+    private static int Compare(Parsed a, Parsed b)
+    {
+        int core = (a.Major, a.Minor, a.Patch).CompareTo((b.Major, b.Minor, b.Patch));
+        if (core != 0)
+        {
+            return core;
+        }
+
+        if (a.Prerelease != b.Prerelease)
+        {
+            return a.Prerelease ? -1 : 1; // 正式版高于预发布版
+        }
+
+        if (!a.Prerelease)
+        {
+            return 0;
+        }
+
+        int dev = a.Dev.CompareTo(b.Dev);
+        return dev != 0 ? dev : 0;
+    }
+
+    private static bool TryParse(string text, out Parsed parsed)
+    {
+        parsed = default;
+        string value = text.Trim().TrimStart('v');
+        int plus = value.IndexOf('+', StringComparison.Ordinal);
+        if (plus >= 0)
+        {
+            value = value[..plus];
+        }
+
+        int dash = value.IndexOf('-', StringComparison.Ordinal);
+        string core = dash < 0 ? value : value[..dash];
+        string suffix = dash < 0 ? "" : value[(dash + 1)..];
+        string[] parts = core.Split('.');
+        if (parts.Length != 3
+            || !int.TryParse(parts[0], NumberStyles.None, CultureInfo.InvariantCulture, out int major)
+            || !int.TryParse(parts[1], NumberStyles.None, CultureInfo.InvariantCulture, out int minor)
+            || !int.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out int patch))
+        {
+            return false;
+        }
+
+        long dev = 0;
+        if (suffix.StartsWith("dev.", StringComparison.Ordinal))
+        {
+            string number = suffix[4..];
+            int end = number.IndexOf('-', StringComparison.Ordinal);
+            if (!long.TryParse(end < 0 ? number : number[..end], NumberStyles.None, CultureInfo.InvariantCulture, out dev))
+            {
+                return false;
+            }
+        }
+
+        parsed = new Parsed(major, minor, patch, suffix.Length > 0, dev, suffix);
+        return true;
+    }
+}

backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs

@@ -3,20 +3,28 @@ using Fleet.Core;
 
 namespace Fleet.Hosting;
 
-/// <summary>Writes logs locally and keeps a bounded copy for the next authenticated heartbeat.</summary>
-public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource
+/// <summary>
+/// Writes logs locally and keeps a bounded ring of recent lines. Heartbeats carry only counters (see <see cref="TakeSummary"/>);
+/// the text leaves the node when the control plane asks for it (see <see cref="Read"/>).
+/// </summary>
+public sealed class BufferedNodeLog : INodeLog, INodeLogSource
 {
-    private const int Capacity = 256;
+    private const int Capacity = 2048;
     private const int MaxFields = 20;
     private const int MaxMessageLength = 512;
     private const int MaxFieldValueLength = 256;
-    private const int MaxBatchBytes = 8 << 10;
+    private const int MaxPageBytes = 96 << 10;
+    private const int MaxSummaryTypes = 8;
 
     private readonly object _gate = new();
-    private readonly Queue<NodeLogRecord> _pending = [];
+    private readonly LinkedList<NodeLogEntry> _ring = [];
+    private readonly Dictionary<(uint Type, string Level), long> _counts = [];
     private readonly INodeLog _local;
     private readonly TimeProvider _time;
-    private long _dropped;
+    private long _lastId;
+    private long _info;
+    private long _warn;
+    private long _error;
 
     public BufferedNodeLog(INodeLog local, TimeProvider time)
     {
@@ -35,50 +43,93 @@ public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource
             safeFields[key.Length <= 64 ? key : key[..64]] = Truncate(safeValue, MaxFieldValueLength);
         }
 
-        var record = new NodeLogRecord(
-            _time.GetUtcNow().ToUnixTimeMilliseconds(),
-            level.ToString().ToLowerInvariant(),
-            Truncate(message, MaxMessageLength),
-            safeFields);
-
+        string levelName = level.ToString().ToLowerInvariant();
+        var record = new NodeLogRecord(_time.GetUtcNow().ToUnixTimeMilliseconds(), levelName, Truncate(message, MaxMessageLength), safeFields);
+        uint type = TypeOf(record.Message);
         lock (_gate)
         {
-            if (_pending.Count == Capacity)
+            _ring.AddLast(new NodeLogEntry(++_lastId, type, record));
+            if (_ring.Count > Capacity)
             {
-                _pending.Dequeue();
-                _dropped++;
+                _ring.RemoveFirst();
             }
 
-            _pending.Enqueue(record);
+            _counts[(type, levelName)] = _counts.GetValueOrDefault((type, levelName)) + 1;
+            switch (level)
+            {
+                case NodeLogLevel.Error:
+                    _error++;
+                    break;
+                case NodeLogLevel.Warn:
+                    _warn++;
+                    break;
+                default:
+                    _info++;
+                    break;
+            }
         }
     }
 
-    public NodeLogBatch TakeBatch(int maxRecords)
+    public NodeLogSummary TakeSummary()
     {
-        int count = Math.Clamp(maxRecords, 0, 32);
         lock (_gate)
         {
-            var records = new List<NodeLogRecord>(Math.Min(count, _pending.Count));
+            var summary = new NodeLogSummary(
+                _lastId, _info, _warn, _error,
+                [.. _counts.OrderByDescending(pair => pair.Value).Take(MaxSummaryTypes).Select(pair => new NodeLogTypeCount(pair.Key.Type, pair.Key.Level, pair.Value))]);
+            _counts.Clear();
+            _info = _warn = _error = 0;
+            return summary;
+        }
+    }
+
+    public NodeLogPage Read(long afterId, int limit, string? minLevel)
+    {
+        int rank = LevelRank(minLevel);
+        lock (_gate)
+        {
+            var items = new List<NodeLogEntry>(Math.Min(Math.Max(limit, 0), 200));
             int bytes = 0;
-            while (records.Count < count && _pending.TryPeek(out NodeLogRecord? record))
+            foreach (NodeLogEntry entry in _ring)
             {
-                int recordBytes = 64 + record.Message.Length + record.Fields.Sum(pair => pair.Key.Length + pair.Value.Length + 8);
-                if (records.Count > 0 && bytes + recordBytes > MaxBatchBytes)
+                if (entry.Id <= afterId || LevelRank(entry.Record.Level) < rank)
+                {
+                    continue;
+                }
+
+                int entryBytes = 64 + entry.Record.Message.Length + entry.Record.Fields.Sum(pair => pair.Key.Length + pair.Value.Length + 8);
+                if (items.Count >= Math.Clamp(limit, 0, 200) || (items.Count > 0 && bytes + entryBytes > MaxPageBytes))
                 {
                     break;
                 }
 
-                _pending.Dequeue();
-                records.Add(record);
-                bytes += recordBytes;
+                items.Add(entry);
+                bytes += entryBytes;
             }
 
-            long dropped = _dropped;
-            _dropped = 0;
-            return new NodeLogBatch(records, dropped);
+            return new NodeLogPage(items, _lastId);
         }
     }
 
+    /// <summary>The line's type: a stable 32-bit FNV-1a hash of its message text, so the same template always maps to the same number.</summary>
+    public static uint TypeOf(string message)
+    {
+        uint hash = 2166136261;
+        foreach (char c in message)
+        {
+            hash = (hash ^ c) * 16777619;
+        }
+
+        return hash;
+    }
+
+    private static int LevelRank(string? level) => level switch
+    {
+        "error" => 2,
+        "warn" => 1,
+        _ => 0,
+    };
+
     private static bool IsSensitiveKey(string key)
     {
         string lower = key.ToLowerInvariant();

backend/fleet/src/Fleet.Link/FleetLink.cs

@@ -52,6 +52,9 @@ public sealed class FleetLinkOptions
 
     public string NodeVersion { get; init; } = "0.0.0";
 
+    /// <summary>服务端至少要是哪个版本（每次握手重新取）；空表示不要求。不够就关掉这条连接，等 <see cref="VersionRetry"/> 再握手，服务端可能正在更新。</summary>
+    public Func<string?>? MinServerVersion { get; init; }
+
     /// <summary>框架之外的能力，例如 <c>moshiro.v1</c> 与 <c>moshiro.kinds:udp_echo,tcp_echo</c>。</summary>
     public IReadOnlyList<string> Capabilities { get; init; } = [];
 
@@ -100,7 +103,11 @@ public sealed class FleetLink
     private volatile RpcSession<LinkContext>? _session;
     private volatile bool _stopped;
 
+    /// <summary>版本不够时重新握手的间隔：固定，不退避（对方更新完马上就能连上）。</summary>
+    public static readonly TimeSpan VersionRetry = TimeSpan.FromSeconds(5);
+
     private static readonly TimeSpan s_tcpTimeout = TimeSpan.FromSeconds(4);
+    private TimeSpan? _fixedRetry;
     private int _failedConnects;
 
     public FleetLink(FleetLinkOptions options, RpcRouter<LinkContext> router, IEnumerable<IStateHandler> handlers, Outbox outbox, TimeProvider time, INodeLog log)
@@ -166,7 +173,8 @@ public sealed class FleetLink
                 return;
             }
 
-            TimeSpan delay = _options.Backoff.Next();
+            TimeSpan delay = _fixedRetry ?? _options.Backoff.Next();
+            _fixedRetry = null;
             _log.Warn("link down, retrying", ("err", reason), ("retry_ms", (long)delay.TotalMilliseconds));
             await Task.Delay(delay, _time, ct);
         }
@@ -263,7 +271,19 @@ public sealed class FleetLink
             Result<HelloResult> helloResult = await session.RequestAsync(FleetMethods.Hello, hello, FleetRpcContext.Default.HelloParams, FleetRpcContext.Default.HelloResult, ct);
             if (!helloResult.IsSuccess)
             {
-                return "hello failed: " + helloResult.Error.Code;
+                if (helloResult.Error.Code == FleetErrors.VersionRequired)
+                {
+                    _fixedRetry = VersionRetry; // 我们的版本低于服务端要求的：等自己更新，每 5 秒再试一次
+                }
+
+                return "hello failed: " + helloResult.Error.Code + (_fixedRetry is null ? "" : ": " + helloResult.Error.Message);
+            }
+
+            string? requiredServer = _options.MinServerVersion?.Invoke();
+            if (NodeVersions.IsBelow(helloResult.Value.ServerVersion, requiredServer))
+            {
+                _fixedRetry = VersionRetry; // 服务端版本低于我们要求的（它可能正在更新）：不发任何东西，5 秒后重新握手
+                return $"server version {helloResult.Value.ServerVersion} is below the required {requiredServer}";
             }
 
             Outbox.ResetSent();

backend/fleet/src/Fleet.Rpc/FleetMethods.cs

@@ -104,6 +104,10 @@ public sealed class HelloResult
     /// <summary>节点时钟 − 服务端时钟（毫秒，未扣除网络往返）。</summary>
     [JsonPropertyName("clock_offset_ms")]
     public long ClockOffsetMs { get; set; }
+
+    /// <summary>服务端的版本：节点要求了最低服务端版本时，用它判断。</summary>
+    [JsonPropertyName("server_version")]
+    public string? ServerVersion { get; set; }
 }
 
 public sealed class HeartbeatParams

backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs

@@ -22,3 +22,12 @@ public sealed class FleetServerOptions
     /// <summary>节点违反协议（方法没有声明、方向、类型或角色不对）的次数上限；超过就断开这条连接。</summary>
     public int MaxViolations { get; set; } = 20;
 }
+
+/// <summary>握手时的版本要求：服务端的版本（告诉节点），以及节点至少要是哪个版本。每次握手都重新取，所以改配置不用重启。</summary>
+public interface IFleetVersionPolicy
+{
+    string? ServerVersion { get; }
+
+    /// <summary>空表示不要求。</summary>
+    string? MinNodeVersion { get; }
+}

backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs

@@ -24,8 +24,10 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier
         IEnumerable<IInbox> inboxes,
         IEnumerable<INodeEventSink> sinks,
         FleetRouter router,
-        IServerIdentity? serverIdentity = null)
+        IServerIdentity? serverIdentity = null,
+        IFleetVersionPolicy? versionPolicy = null)
     {
+        VersionPolicy = versionPolicy;
         ServerIdentity = serverIdentity;
         Store = store;
         Options = options;
@@ -43,6 +45,8 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier
     /// <summary>服务器的身份密钥；没有时节点连接用旧的明文认证。</summary>
     internal IServerIdentity? ServerIdentity { get; }
 
+    internal IFleetVersionPolicy? VersionPolicy { get; }
+
     internal FleetServerOptions Options { get; }
 
     internal TimeProvider Time { get; }

backend/fleet/src/Fleet.Server.Link/FleetLog.cs

@@ -10,6 +10,9 @@ internal static partial class FleetLog
     [LoggerMessage(Level = LogLevel.Information, Message = "node disconnected node={Node} reason={Reason}")]
     public static partial void Disconnected(ILogger logger, string node, string reason);
 
+    [LoggerMessage(Level = LogLevel.Warning, Message = "node below the required version node={Node} version={Version} required={Required}")]
+    public static partial void VersionTooOld(ILogger logger, string node, string version, string required);
+
     [LoggerMessage(Level = LogLevel.Warning, Message = "node auth rejected err={Err}")]
     public static partial void AuthRejected(ILogger logger, string err);
 

backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs

@@ -110,7 +110,8 @@ public static class FleetServerExtensions
             provider.GetServices<IInbox>(),
             provider.GetServices<INodeEventSink>(),
             provider.GetRequiredService<FleetRouter>(),
-            provider.GetService<IServerIdentity>()));
+            provider.GetService<IServerIdentity>(),
+            provider.GetService<IFleetVersionPolicy>()));
         services.AddSingleton<IFleetConnections>(provider => provider.GetRequiredService<FleetLinkHost>());
         services.AddSingleton<IStateNotifier>(provider => provider.GetRequiredService<FleetLinkHost>());
         return builder;

backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs

@@ -151,6 +151,20 @@ internal sealed partial class NodeConnection
             return;
         }
 
+        string? minVersion = _host.VersionPolicy?.MinNodeVersion;
+        if (NodeVersions.IsBelow(hello.NodeVersion, minVersion))
+        {
+            // 版本不够：不登记在线，告诉节点要求的版本，关闭连接。节点等几秒（它可能正在更新）再握手。
+            FleetLog.VersionTooOld(_host.Logger, _node!.Id.Value, hello.NodeVersion, minVersion!);
+            if (message.Id is not null)
+            {
+                await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.VersionRequired, $"node version {hello.NodeVersion} is below the required {minVersion}")), ct);
+            }
+
+            await CloseAsync("version required");
+            return;
+        }
+
         long nowMs = _host.Time.GetUtcNow().ToUnixTimeMilliseconds();
         _session = _session! with
         {
@@ -169,7 +183,7 @@ internal sealed partial class NodeConnection
             await SendAsync(
                 RpcMessage.Success(
                     message.Id.Value,
-                    new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs },
+                    new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs, ServerVersion = _host.VersionPolicy?.ServerVersion },
                     FleetRpcContext.Default.HelloResult),
                 ct);
         }

backend/fleet/tests/Fleet.Tests/BufferedNodeLogTests.cs

@@ -0,0 +1,65 @@
+using Fleet.Core;
+using Fleet.Hosting;
+
+namespace Fleet.Tests;
+
+public sealed class BufferedNodeLogTests
+{
+    [Fact]
+    public void Summary_CountsByLevelAndTypeWithoutText_AndResetsAfterEachTake()
+    {
+        var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System);
+        for (int i = 0; i < 5; i++)
+        {
+            log.Warn("peer control handshake failed", ("attempt", i));
+        }
+
+        log.Info("link up");
+        log.Error("MsQuic QUIC transport shutdown");
+
+        NodeLogSummary first = log.TakeSummary();
+        Assert.Equal((1L, 5L, 1L), (first.Info, first.Warn, first.Error));
+        Assert.Equal(7, first.LastId);
+        NodeLogTypeCount top = first.Types[0];
+        Assert.Equal((BufferedNodeLog.TypeOf("peer control handshake failed"), "warn", 5L), (top.Type, top.Level, top.Count));
+
+        NodeLogSummary second = log.TakeSummary();
+        Assert.Equal((0L, 0L, 0L), (second.Info, second.Warn, second.Error));
+        Assert.Empty(second.Types);
+        Assert.Equal(7, second.LastId);
+    }
+
+    [Fact]
+    public void Read_ReturnsLinesAfterAnIdAtOrAboveALevel_OldestFirst_AndTheRingKeepsTheNewest()
+    {
+        var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System);
+        log.Info("one");
+        log.Warn("two");
+        log.Error("three");
+        log.Info("four");
+
+        Assert.Equal(["two", "three"], log.Read(1, 2, null).Items.Select(e => e.Record.Message));
+        Assert.Equal(["three"], log.Read(0, 10, "error").Items.Select(e => e.Record.Message));
+        Assert.Equal(["two", "three"], log.Read(0, 10, "warn").Items.Select(e => e.Record.Message));
+
+        for (int i = 0; i < 3000; i++)
+        {
+            log.Info("filler", ("n", i));
+        }
+
+        NodeLogPage page = log.Read(0, 200, null);
+        Assert.Equal(200, page.Items.Count);
+        Assert.True(page.Items[0].Id > 4); // 最早的几条已经被环挤掉
+        Assert.Equal(3004, page.LastId);
+    }
+
+    [Fact]
+    public void SensitiveFieldsAreRedacted()
+    {
+        var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System);
+        log.Info("auth", ("token", "abc"), ("agent", "agt_x"));
+        NodeLogEntry entry = Assert.Single(log.Read(0, 10, null).Items);
+        Assert.Equal("[redacted]", entry.Record.Fields["token"]);
+        Assert.Equal("agt_x", entry.Record.Fields["agent"]);
+    }
+}

backend/fleet/tests/Fleet.Tests/LinkTests.cs

@@ -373,9 +373,9 @@ public sealed class LinkTests
             Assert.Equal("dlr_1", server.PeerId);
             Assert.Equal(client.Credentials.Key, server.Credentials.Key);
             Assert.Equal(client.Credentials.SessionId, server.Credentials.SessionId);
-            NodeLogBatch batch = logs.TakeBatch(32);
-            Assert.Equal(new[] { "tcp_connect", "tls", "peer_auth" }, batch.Records.Select(x => x.Fields["phase"]));
-            Assert.All(batch.Records, entry => Assert.Equal("test-attempt", entry.Fields["attempt_id"]));
+            NodeLogPage page = logs.Read(0, 32, null);
+            Assert.Equal(new[] { "tcp_connect", "tls", "peer_auth" }, page.Items.Select(x => x.Record.Fields["phase"]));
+            Assert.All(page.Items, entry => Assert.Equal("test-attempt", entry.Record.Fields["attempt_id"]));
         }
     }
 
@@ -391,7 +391,7 @@ public sealed class LinkTests
         var logs = new Fleet.Hosting.BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System);
         InvalidOperationException unknown = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
             await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, dialer, "dlr_unknown", null, 1 << 20, TimeProvider.System, logs, default));
-        NodeLogRecord failure = Assert.Single(logs.TakeBatch(32).Records, entry => entry.Message == "peer control handshake failed");
+        NodeLogRecord failure = Assert.Single(logs.Read(0, 32, null).Items, entry => entry.Record.Message == "peer control handshake failed").Record;
         Assert.Equal("peer_auth", failure.Fields["phase"]);
         Assert.Equal("InvalidOperationException", failure.Fields["exception"]);
         InvalidOperationException forged = await Assert.ThrowsAsync<InvalidOperationException>(async () =>

backend/fleet/tests/Fleet.Tests/NodeVersionsTests.cs

@@ -0,0 +1,21 @@
+using Fleet.Core;
+
+namespace Fleet.Tests;
+
+public sealed class NodeVersionsTests
+{
+    [Theory]
+    [InlineData("0.4.2-dev.73-5245f03", "0.4.2-dev.74", true)]
+    [InlineData("0.4.2-dev.73-5245f03", "0.4.2-dev.73-abcdef0", false)] // 哈希不参与比较
+    [InlineData("0.4.2-dev.100-aaaaaaa", "0.4.2-dev.99", false)] // 序号按数字比，不是按字符串
+    [InlineData("0.4.2-dev.999", "0.4.2", true)] // 正式版高于预发布版
+    [InlineData("0.4.2", "0.4.2-dev.5", false)]
+    [InlineData("0.4.1", "0.4.2-dev.1", true)]
+    [InlineData("v0.5.0", "0.4.9", false)]
+    [InlineData("0.4.2-dev.73-5245f03", "", false)] // 没有要求
+    [InlineData("", "0.4.2", false)] // 认不出来的版本号不拒绝
+    [InlineData("not-a-version", "0.4.2", false)]
+    [InlineData("0.4.2", "garbage", false)]
+    public void IsBelow_ComparesCoreThenReleaseThenTheDevCounter(string actual, string required, bool expected) =>
+        Assert.Equal(expected, NodeVersions.IsBelow(actual, required));
+}

backend/openapi/v1.json

@@ -1897,7 +1897,7 @@
         "tags": [
           "admin-nodes"
         ],
-        "summary": "Recent runtime logs of a node",
+        "summary": "Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept",
         "operationId": "node_runtime_logs",
         "parameters": [
           {
@@ -9859,4 +9859,4 @@
       "name": "agent"
     }
   ]
-}
+}

backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs

@@ -1,4 +1,5 @@
 using System.Net;
+using Fleet.Core;
 using MoShiro.Protocol;
 
 namespace MoShiro.Measurement;
@@ -41,6 +42,76 @@ public abstract class BlockBase : IBlock
         Env.Fragments!.Ship(Spec.KindId, Spec.Scope, streams, parameters ?? Spec.Params.Values, Env.Clock.UtcMsAt(first), (int)((last - first) / 1000), Spec.Port);
     }
 
+    /// <summary>
+    /// 按自适应的节奏发包：下一个包的时刻按绝对时间排（不累积误差），每秒评估一次状态，状态变化记进日志（供审计）。
+    /// 落后了就从现在重新排，不补发。
+    /// </summary>
+    protected async Task PaceAsync(AdaptiveRate rate, Func<ValueTask> sendOne, CancellationToken ct)
+    {
+        long due = Env.Clock.NowUs();
+        long lastEvaluation = due;
+        while (!ct.IsCancellationRequested)
+        {
+            due += (long)(rate.NextDelay().TotalMilliseconds * 1000);
+            long wait = due - Env.Clock.NowUs();
+            if (wait > 0)
+            {
+                await Task.Delay(TimeSpan.FromMicroseconds(wait), Env.Time, ct);
+            }
+            else
+            {
+                due = Env.Clock.NowUs();
+            }
+
+            await sendOne();
+            long now = Env.Clock.NowUs();
+            if (now - lastEvaluation >= 1_000_000 || rate.State == ProbeState.Dead)
+            {
+                lastEvaluation = now;
+                if (rate.Evaluate(now) is { } change)
+                {
+                    LogTransition(change);
+                }
+            }
+        }
+    }
+
+    /// <summary>读这个任务的自适应频率参数（<c>udp_echo</c> 和 <c>tcp_echo</c> 共用）。</summary>
+    protected AdaptiveRate? CreateAdaptiveRate()
+    {
+        if (!Spec.Bool("adaptive"))
+        {
+            return null;
+        }
+
+        double max = Math.Max(Spec.Int("rate_max_hz"), Spec.Int("rate_min_hz"));
+        return new AdaptiveRate(new AdaptiveRateOptions
+        {
+            MinHz = Spec.Int("rate_min_hz"),
+            MaxHz = max,
+            TauPct = Spec.Number("tau_pct"),
+            RttTolPct = Spec.Number("rtt_tol_pct"),
+            Prior = Env.LossPrior,
+        });
+    }
+
+    private void LogTransition(ProbeTransition change)
+    {
+        var fields = new (string, object?)[]
+        {
+            ("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
+            ("reason", change.Reason), ("loss_pct", Math.Round(change.LossPct, 3)), ("sent", change.Sent), ("lost", change.Lost),
+        };
+        if (change.To == ProbeState.Dead)
+        {
+            Env.Log.Warn("probe rate changed", fields);
+        }
+        else
+        {
+            Env.Log.Info("probe rate changed", fields);
+        }
+    }
+
     /// <summary>什么也不做、一直等到取消（监听方里不需要自己运行逻辑的积木）。</summary>
     protected async Task WaitUntilCancelledAsync(CancellationToken ct)
     {

backend/shared/MoShiro.Measurement/Blocks/Echo.cs

@@ -23,6 +23,7 @@ internal sealed class EchoRows
 
     private readonly DriverLog _log;
     private readonly int _size;
+    private readonly int _w;
     private readonly int _t1;
     private readonly int _t4;
     private readonly long _graceUs;
@@ -36,26 +37,36 @@ internal sealed class EchoRows
         StreamSpec spec = kind.FindStream("msg")!;
         _log = new DriverLog(spec, Side.Dialer, capacity);
         _size = _log.Store.ColumnIndex("size");
+        _w = _log.Store.ColumnIndex("w");
         _t1 = _log.Store.ColumnIndex("t1");
         _t4 = _log.Store.ColumnIndex("t4");
     }
 
     public long Count => _log.Count - Interlocked.Read(ref _discarded);
 
-    public uint Add(int size, long t1Us)
+    /// <summary><paramref name="weight"/>：这个包代表多少个基础网格间隔（频率变化时分析按它累计，丢包率和分位数仍是按时间平均的）。</summary>
+    public uint Add(int size, long t1Us, int weight = 1)
     {
         long key = _log.Store.Append();
         _log.Store.Set(key, _size, size);
+        _log.Store.Set(key, _w, weight);
         _log.Store.Set(key, _t1, t1Us);
         return (uint)key;
     }
 
-    public void Reply(uint seq, long t4Us)
+    public bool Reply(uint seq, long t4Us) => Reply(seq, t4Us, out _);
+
+    /// <summary>回填 t4；第一次回复才算，<paramref name="t1Us"/> 是这个包的发出时间。</summary>
+    public bool Reply(uint seq, long t4Us, out long t1Us)
     {
-        if (_log.Store.Contains(seq) && !_log.Store.IsSet(seq, _t4))
+        t1Us = 0;
+        if (_log.Store.Contains(seq) && !_log.Store.IsSet(seq, _t4) && _log.Store.TryGet(seq, _t1, out t1Us))
         {
             _log.Store.Set(seq, _t4, t4Us);
+            return true;
         }
+
+        return false;
     }
 
     public bool Discard(uint seq)
@@ -83,6 +94,7 @@ public sealed class DatagramEchoDialer : BlockBase
     private readonly EchoRows _rows;
     private readonly object _gate = new();
     private long _nextSendFailureWarningUs;
+    private AdaptiveRate? _rate;
 
     public DatagramEchoDialer(BlockEnv env, BlockSpec spec)
         : base(env, spec)
@@ -100,29 +112,50 @@ public sealed class DatagramEchoDialer : BlockBase
         IPEndPoint target = PeerEndpoint(Spec.Port);
         int size = (int)Spec.Int("size");
         byte[] packet = new byte[size];
+        _rate = CreateAdaptiveRate();
+
+        void SendOne()
+        {
+            long t1 = Env.Clock.NowUs();
+            lock (_gate)
+            {
+                uint seq = _rows.Add(size, t1, _rate?.Weight ?? 1);
+                MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
+                if (datagrams.Send(session, target, packet))
+                {
+                    _rate?.OnSend(t1);
+                    return;
+                }
+
+                _rows.Discard(seq);
+                if (t1 >= _nextSendFailureWarningUs)
+                {
+                    Env.Log.Warn("QUIC echo datagram was not queued; omitted from loss samples",
+                        ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
+                    _nextSendFailureWarningUs = t1 + 5_000_000;
+                }
+            }
+        }
+
         datagrams.Register(Spec.StreamId, OnPacket);
         try
         {
-            using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
-            while (await timer.WaitForNextTickAsync(ct))
+            if (_rate is null)
             {
-                long t1 = Env.Clock.NowUs();
-                lock (_gate)
+                using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
+                while (await timer.WaitForNextTickAsync(ct))
                 {
-                    uint seq = _rows.Add(size, t1);
-                    MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
-                    if (!datagrams.Send(session, target, packet))
-                    {
-                        _rows.Discard(seq);
-                        if (t1 >= _nextSendFailureWarningUs)
-                        {
-                            Env.Log.Warn("QUIC echo datagram was not queued; omitted from loss samples",
-                                ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
-                            _nextSendFailureWarningUs = t1 + 5_000_000;
-                        }
-                    }
+                    SendOne();
                 }
             }
+            else
+            {
+                await PaceAsync(_rate, () =>
+                {
+                    SendOne();
+                    return ValueTask.CompletedTask;
+                }, ct);
+            }
         }
         catch (OperationCanceledException)
         {
@@ -157,7 +190,10 @@ public sealed class DatagramEchoDialer : BlockBase
         {
             lock (_gate)
             {
-                _rows.Reply(delivery.Header.Seq, delivery.ReceiveUs);
+                if (_rows.Reply(delivery.Header.Seq, delivery.ReceiveUs, out long t1))
+                {
+                    _rate?.OnReply(t1, delivery.ReceiveUs, (delivery.ReceiveUs - t1) / 1000.0);
+                }
             }
         }
     }
@@ -211,6 +247,7 @@ public sealed class TcpEchoDialer : BlockBase
 {
     private readonly EchoRows _rows;
     private readonly TcpInfoRecorder _info;
+    private AdaptiveRate? _rate;
 
     public TcpEchoDialer(BlockEnv env, BlockSpec spec)
         : base(env, spec)
@@ -224,12 +261,14 @@ public sealed class TcpEchoDialer : BlockBase
     public override async Task RunAsync(CancellationToken ct)
     {
         var backoff = new Fleet.Core.Backoff(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
+        _rate = CreateAdaptiveRate();
         while (!ct.IsCancellationRequested)
         {
             try
             {
                 using TcpLink link = await TcpLink.OpenAsync(Env, Spec.Port, Spec.Bool("tls"), Spec.StreamId, sendAuth: true, ct);
                 backoff.Reset();
+                _rate?.Reconnected(Env.Clock.NowUs());
                 await ExchangeAsync(link, ct);
             }
             catch (OperationCanceledException) when (ct.IsCancellationRequested)
@@ -300,16 +339,29 @@ public sealed class TcpEchoDialer : BlockBase
         int min = (int)Spec.Int("min");
         int max = (int)Spec.Int("max");
         byte[] body = new byte[Math.Max(max, min)];
-        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
-        while (await timer.WaitForNextTickAsync(ct))
+
+        async ValueTask SendOneAsync()
         {
             int size = Math.Max(TcpWire.MinEchoBody, max > min ? Random.Shared.Next(min, max + 1) : min);
             long t1 = Env.Clock.NowUs();
-            uint seq = _rows.Add(size, t1);
+            uint seq = _rows.Add(size, t1, _rate?.Weight ?? 1);
             body.AsSpan(0, size).Clear();
             TcpWire.WriteEchoBody(body, seq, t1);
             await TcpWire.WriteMessageAsync(stream, body.AsMemory(0, size), ct);
             await stream.FlushAsync(ct);
+            _rate?.OnSend(t1);
+        }
+
+        if (_rate is not null)
+        {
+            await PaceAsync(_rate, SendOneAsync, ct);
+            return;
+        }
+
+        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
+        while (await timer.WaitForNextTickAsync(ct))
+        {
+            await SendOneAsync();
         }
     }
 
@@ -324,9 +376,9 @@ public sealed class TcpEchoDialer : BlockBase
             }
 
             long now = Env.Clock.NowUs();
-            if (TcpWire.TryReadEchoBody(reply, out uint seq, out _))
+            if (TcpWire.TryReadEchoBody(reply, out uint seq, out _) && _rows.Reply(seq, now, out long t1))
             {
-                _rows.Reply(seq, now);
+                _rate?.OnReply(t1, now, (now - t1) / 1000.0);
             }
         }
     }

backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs

@@ -0,0 +1,449 @@
+namespace MoShiro.Measurement;
+
+/// <summary>探测节奏的状态。<see cref="Dead"/> 是线路几乎全丢：不再按正常频率白白发包，指数退避着探测，等到有回包再恢复。</summary>
+public enum ProbeState : byte
+{
+    Quiet = 0,
+    Watch = 1,
+    Active = 2,
+    Dead = 3,
+}
+
+/// <summary>自适应探测频率的参数。频率范围、精度目标和各状态的保持时间都在这里。</summary>
+public sealed record AdaptiveRateOptions
+{
+    /// <summary>平稳时的最低频率（Hz）。</summary>
+    public double MinHz { get; init; } = 2;
+
+    /// <summary>
+    /// 最高频率（Hz），也是权重的参考网格：每个探测包的权重是它代表的时间除以 <c>1/MaxHz</c>，
+    /// 分析时丢包率和 RTT 分位数按权重累计，所以频率怎么变，得到的仍是按时间平均的值，不会偏向探测更密的时段。
+    /// </summary>
+    public double MaxHz { get; init; } = 10;
+
+    /// <summary>平稳时 RTT 分位数（p99）的 95% 置信区间半宽容差：取 max(下限毫秒, 百分比 × p99)。样本多到超出这个精度才会降频。</summary>
+    public double RttTolPct { get; init; } = 2;
+
+    public double RttTolFloorMs { get; init; } = 0.3;
+
+    /// <summary>平稳线路要能证明的丢包上界（百分比，95% 置信）。三法则：零丢包需要 3 / τ 个样本。</summary>
+    public double TauPct { get; init; } = 0.2;
+
+    /// <summary>证明窗口（秒）：平稳频率 = 3 / (τ × 窗口)。</summary>
+    public double CertifySeconds { get; init; } = 300;
+
+    /// <summary>历史上一直平稳的时段可以拉长证明窗口，同样的精度用更少的包。</summary>
+    public double QuietCertifySeconds { get; init; } = 900;
+
+    /// <summary>回包等多久算丢（秒）。</summary>
+    public double GraceSeconds { get; init; } = 3;
+
+    /// <summary>刚启动没有历史：先按 Watch 频率跑这么久（秒）。</summary>
+    public double WarmupSeconds { get; init; } = 300;
+
+    /// <summary>一次丢包事件之后，Watch 至少保持多久（秒）。</summary>
+    public double WatchHoldSeconds { get; init; } = 1800;
+
+    /// <summary>进入 Active 之后，没有新的丢包事件要保持多久才降到 Watch（秒）。</summary>
+    public double ActiveHoldSeconds { get; init; } = 900;
+
+    /// <summary>证明窗口内有这么多个不同的秒出现丢包，就算持续丢包，升到 Active。</summary>
+    public int ActiveLossCount { get; init; } = 3;
+
+    /// <summary>判断线路全丢的窗口（秒）和最少样本数、丢包比例。</summary>
+    public double DeadWindowSeconds { get; init; } = 10;
+
+    public int DeadMinSent { get; init; } = 10;
+
+    public double DeadLossRatio { get; init; } = 0.95;
+
+    /// <summary>Dead 状态的探测间隔：从 1 秒起翻倍，最长这么久（秒），带 ±20% 抖动。</summary>
+    public double DeadMaxIntervalSeconds { get; init; } = 30;
+
+    /// <summary>RTT 比长期水平高出这个倍数（再加下面的绝对值）持续几秒，就当作链路在变差，至少升到 Watch。</summary>
+    public double RttFactor { get; init; } = 1.5;
+
+    public double RttFloorMs { get; init; } = 5;
+
+    /// <summary>先验：时间 → 这个时刻丢包的历史概率（0 到 1）；null 表示没有历史。</summary>
+    public Func<long, double?>? Prior { get; init; }
+
+    /// <summary>先验里的"丢包高发"阈值：达到它在到来之前 <see cref="PriorLeadSeconds"/> 提前预热。</summary>
+    public double PriorWatchAt { get; init; } = 0.15;
+
+    public double PriorActiveAt { get; init; } = 0.5;
+
+    public double PriorLeadSeconds { get; init; } = 1800;
+}
+
+/// <summary>状态或频率变了：<see cref="Reason"/> 是给日志和审计看的固定短语。</summary>
+public sealed record ProbeTransition(ProbeState From, ProbeState To, double Hz, string Reason, double LossPct, int Sent, int Lost);
+
+/// <summary>
+/// 自适应探测频率：按"要证明的精度"决定频率，丢包多了自动升频，线路全丢就指数退避。
+/// 平稳线路只需要能证明丢包低于 τ 的样本数（三法则 N = 3/τ），有丢包的线路用满频率保精度和丢包的细节（连续丢包、乱序）。
+/// 状态靠丢包事件的频率升降，带滞回，不会在阈值附近来回跳；时间画像可以让高发时段提前预热。
+/// 纯逻辑：时间由调用方传入，不依赖时钟，方便测试。
+/// </summary>
+public sealed class AdaptiveRate
+{
+    private const long UsPerSecond = 1_000_000;
+    private const int Buckets = 1200; // 20 分钟，够覆盖最长的统计窗口
+
+    private readonly AdaptiveRateOptions _options;
+    private readonly Random _random;
+    private readonly object _gate = new();
+    private readonly long[] _second = new long[Buckets];
+    private readonly int[] _sent = new int[Buckets];
+    private readonly int[] _replied = new int[Buckets];
+
+    private long _startUs = -1;
+    private long _lastLossUs = long.MinValue;
+    private long _activeUntilUs = long.MinValue;
+    private long _lastReplyUs = long.MinValue;
+    private long _deadSinceUs = long.MinValue;
+    private int _deadProbes;
+    private readonly Queue<(long Second, double Ms)> _rtts = new();
+    private long _lastGuardSecond = long.MinValue;
+    private double _guardHz = double.NaN;
+    private TimeSpan _lastDelay = TimeSpan.Zero;
+    private double _rttFast = double.NaN;
+    private double _rttSlow = double.NaN;
+    private int _rttHighSeconds;
+    private long _lastRttSecond = long.MinValue;
+
+    public AdaptiveRate(AdaptiveRateOptions options, Random? random = null)
+    {
+        _options = options;
+        _random = random ?? Random.Shared;
+        State = ProbeState.Watch;
+        Hz = _options.MaxHz;
+    }
+
+    public ProbeState State { get; private set; }
+
+    /// <summary>当前的探测频率（Hz）。Dead 时是退避后的近似频率。</summary>
+    public double Hz { get; private set; }
+
+    /// <summary>发出一个探测包。</summary>
+    public void OnSend(long nowUs)
+    {
+        lock (_gate)
+        {
+            if (_startUs < 0)
+            {
+                _startUs = nowUs;
+            }
+
+            long second = nowUs / UsPerSecond;
+            int slot = Slot(second);
+            if (_second[slot] != second)
+            {
+                _second[slot] = second;
+                _sent[slot] = 0;
+                _replied[slot] = 0;
+            }
+
+            _sent[slot]++;
+            if (State == ProbeState.Dead)
+            {
+                _deadProbes++;
+            }
+        }
+    }
+
+    /// <summary>收到回包：<paramref name="sentUs"/> 是它发出的时间，<paramref name="rttMs"/> 是往返时延。</summary>
+    public void OnReply(long sentUs, long nowUs, double rttMs)
+    {
+        lock (_gate)
+        {
+            long second = sentUs / UsPerSecond;
+            int slot = Slot(second);
+            if (_second[slot] == second && _replied[slot] < _sent[slot])
+            {
+                _replied[slot]++;
+            }
+
+            if (double.IsNaN(_rttSlow))
+            {
+                _rttSlow = _rttFast = rttMs;
+            }
+            else
+            {
+                // 时间常数：快的 5 秒，慢的 15 分钟（按时间而不是按包数，频率变了也一样）
+                double dt = Math.Max(0, (nowUs - Math.Max(_lastReplyUs, 0)) / (double)UsPerSecond);
+                _rttFast += (1 - Math.Exp(-dt / 5)) * (rttMs - _rttFast);
+                _rttSlow += (1 - Math.Exp(-dt / 900)) * (rttMs - _rttSlow);
+            }
+
+            _lastReplyUs = nowUs;
+            _rtts.Enqueue((nowUs / UsPerSecond, rttMs));
+            while (_rtts.Count > 6000)
+            {
+                _rtts.Dequeue();
+            }
+        }
+    }
+
+    /// <summary>连接（重新）建立了：之前线路不通退避着的话，马上恢复探测，不用等退避间隔。</summary>
+    public void Reconnected(long nowUs)
+    {
+        lock (_gate)
+        {
+            if (State == ProbeState.Dead)
+            {
+                _lastReplyUs = nowUs;
+            }
+        }
+    }
+
+    /// <summary>距离下一个探测包还有多久（带抖动，平均值等于频率的倒数）。</summary>
+    public TimeSpan NextDelay()
+    {
+        lock (_gate)
+        {
+            if (State == ProbeState.Dead)
+            {
+                double seconds = Math.Min(_options.DeadMaxIntervalSeconds, Math.Pow(2, Math.Min(_deadProbes, 10)));
+                return _lastDelay = TimeSpan.FromSeconds(seconds * (0.8 + 0.4 * _random.NextDouble()));
+            }
+
+            // 平稳时间隔精确（不加抖动）：发包时刻规整，列块里的时间戳差分更小；抖动只用在退避里避免多条线路同步重试
+            return _lastDelay = TimeSpan.FromSeconds(1.0 / Hz);
+        }
+    }
+
+    /// <summary>刚刚等过的那个间隔所代表的权重（以 1/MaxHz 为单位）：下一个发出的包用它。平稳时是 MaxHz/Hz，退避时是几十到几百。</summary>
+    public int Weight
+    {
+        get
+        {
+            lock (_gate)
+            {
+                return (int)Math.Clamp(Math.Round(_lastDelay.TotalSeconds * _options.MaxHz), 1, 65535);
+            }
+        }
+    }
+
+    /// <summary>重新评估状态，大约每秒调用一次；状态或频率变了就返回变化。</summary>
+    public ProbeTransition? Evaluate(long nowUs)
+    {
+        lock (_gate)
+        {
+            if (_startUs < 0)
+            {
+                return null;
+            }
+
+            long finalSecond = nowUs / UsPerSecond - (long)Math.Ceiling(_options.GraceSeconds);
+            (int sent, int replied, int _) shortWindow = Window(finalSecond, (int)_options.DeadWindowSeconds);
+            (int sent, int replied, int lossSeconds) certify = Window(finalSecond, (int)_options.CertifySeconds);
+            int lostCertify = certify.sent - certify.replied;
+            double lossPct = certify.sent == 0 ? 0 : 100.0 * lostCertify / certify.sent;
+
+            // 最近一次有丢包的时间：用最近 CertifySeconds 里最晚一个有丢包的整秒
+            for (long second = finalSecond; second > finalSecond - (long)_options.CertifySeconds; second--)
+            {
+                int slot = Slot(second);
+                if (_second[slot] == second && _sent[slot] > _replied[slot])
+                {
+                    _lastLossUs = Math.Max(_lastLossUs, second * UsPerSecond);
+                    break;
+                }
+            }
+
+            UpdateRttTrend(nowUs);
+
+            ProbeState next;
+            string reason;
+            if (State == ProbeState.Dead)
+            {
+                if (_lastReplyUs > _deadSinceUs)
+                {
+                    next = ProbeState.Watch;
+                    reason = "reply received";
+                    _deadProbes = 0;
+                    _lastLossUs = nowUs;
+                }
+                else
+                {
+                    return null;
+                }
+            }
+            else if (shortWindow.sent >= _options.DeadMinSent && shortWindow.replied <= shortWindow.sent * (1 - _options.DeadLossRatio))
+            {
+                next = ProbeState.Dead;
+                reason = "line unreachable";
+                _deadSinceUs = nowUs;
+                _deadProbes = 0;
+            }
+            else
+            {
+                (next, reason) = Decide(nowUs, certify.lossSeconds, lostCertify, lossPct);
+            }
+
+            UpdateGuard(nowUs);
+            double hz = next switch
+            {
+                ProbeState.Quiet => Quantize(Math.Max(QuietHz(QuietWindowSeconds(nowUs)), double.IsNaN(_guardHz) ? _options.MaxHz : _guardHz)),
+                ProbeState.Watch or ProbeState.Active => _options.MaxHz,
+                _ => Math.Max(_options.MinHz / 4, 1.0 / _options.DeadMaxIntervalSeconds),
+            };
+            hz = Math.Min(_options.MaxHz, hz);
+
+            if (next == State && Math.Abs(hz - Hz) / Hz < 0.05)
+            {
+                return null;
+            }
+
+            ProbeTransition result = new(State, next, hz, reason, lossPct, certify.sent, lostCertify);
+            State = next;
+            Hz = hz;
+            return result;
+        }
+    }
+
+    private (ProbeState State, string Reason) Decide(long nowUs, int lossSeconds, int lostCertify, double lossPct)
+    {
+        ProbeState state = ProbeState.Quiet;
+        string reason = "quiet";
+
+        double sinceLoss = _lastLossUs == long.MinValue ? double.PositiveInfinity : (nowUs - _lastLossUs) / (double)UsPerSecond;
+        if (lossSeconds >= _options.ActiveLossCount || (lossPct >= _options.TauPct * 2 && lossSeconds >= 2))
+        {
+            _activeUntilUs = nowUs + (long)(_options.ActiveHoldSeconds * UsPerSecond);
+        }
+
+        if (nowUs < _activeUntilUs)
+        {
+            return (ProbeState.Active, "sustained loss");
+        }
+
+        if (sinceLoss <= _options.WatchHoldSeconds)
+        {
+            state = ProbeState.Watch;
+            reason = "recent loss";
+        }
+
+        if (_rttHighSeconds >= 3)
+        {
+            state = ProbeState.Watch;
+            reason = "rtt above baseline";
+        }
+
+        double? prior = _options.Prior?.Invoke(nowUs);
+        double? ahead = _options.Prior?.Invoke(nowUs + (long)(_options.PriorLeadSeconds * UsPerSecond));
+        double expected = Math.Max(prior ?? 0, ahead ?? 0);
+        if (expected >= _options.PriorActiveAt)
+        {
+            return (ProbeState.Active, "expected peak");
+        }
+
+        if (expected >= _options.PriorWatchAt && state == ProbeState.Quiet)
+        {
+            state = ProbeState.Watch;
+            reason = "expected loss window";
+        }
+
+        if (state == ProbeState.Quiet && _options.Prior is null && nowUs - _startUs < (long)(_options.WarmupSeconds * UsPerSecond))
+        {
+            state = ProbeState.Watch;
+            reason = "warm-up";
+        }
+
+        return (state, reason);
+    }
+
+    /// <summary>历史上这个时段（现在和将要到来的一段时间）都平稳时，证明窗口可以更长，频率更低。</summary>
+    private double QuietWindowSeconds(long nowUs)
+    {
+        if (_options.Prior is null)
+        {
+            return _options.CertifySeconds;
+        }
+
+        double now = _options.Prior(nowUs) ?? 1;
+        double ahead = _options.Prior(nowUs + (long)(_options.PriorLeadSeconds * UsPerSecond)) ?? 1;
+        return Math.Max(now, ahead) <= 0.02 ? _options.QuietCertifySeconds : _options.CertifySeconds;
+    }
+
+    /// <summary>频率取 MaxHz/k（k 为整数）：每个包的权重正好是整数 k，分析时没有舍入误差。</summary>
+    private double Quantize(double hz)
+    {
+        double k = Math.Max(1, Math.Floor(_options.MaxHz / Math.Max(hz, 1e-9)));
+        return _options.MaxHz / k;
+    }
+
+    /// <summary>
+    /// 分位数精度保护：用最近一个证明窗口里的 RTT 样本估计 p99 的 95% 置信区间（次序统计量），
+    /// 半宽超过容差就升频，远小于容差才降频（单步最多减半），使平稳线路的 RTT 分位数始终保持在设定的精度内。
+    /// </summary>
+    private void UpdateGuard(long nowUs)
+    {
+        long second = nowUs / UsPerSecond;
+        if (_lastGuardSecond != long.MinValue && second - _lastGuardSecond < 10)
+        {
+            return;
+        }
+
+        _lastGuardSecond = second;
+        long from = second - (long)_options.CertifySeconds;
+        double[] window = [.. _rtts.Where(r => r.Second >= from).Select(r => r.Ms).Order()];
+        if (window.Length < 200)
+        {
+            _guardHz = double.NaN; // 样本不够估计精度：按最高频率
+            return;
+        }
+
+        int n = window.Length;
+        double rank = n * 0.99;
+        double half = 1.96 * Math.Sqrt(n * 0.99 * 0.01);
+        int lo = Math.Clamp((int)Math.Floor(rank - half), 0, n - 1);
+        int hi = Math.Clamp((int)Math.Ceiling(rank + half), 0, n - 1);
+        double halfWidth = (window[hi] - window[lo]) / 2;
+        double tolerance = Math.Max(_options.RttTolFloorMs, _options.RttTolPct / 100.0 * window[Math.Min(n - 1, (int)rank)]);
+        double ratio = Math.Clamp(halfWidth / tolerance, 0.7, 2.0); // 单步变化有限，避免来回跳
+        double current = Hz;
+        _guardHz = Math.Clamp(current * ratio * ratio, _options.MinHz, _options.MaxHz);
+    }
+
+    private double QuietHz(double windowSeconds) => Math.Min(_options.MaxHz, Math.Max(_options.MinHz, 3.0 / (_options.TauPct / 100.0 * windowSeconds)));
+
+    /// <summary>窗口内发出、回来的包数，以及有丢包的整秒数（一次连续丢包只算一个"事件"，不会因为一次短暂中断就判成持续丢包）。</summary>
+    private (int Sent, int Replied, int LossSeconds) Window(long finalSecond, int seconds)
+    {
+        int sent = 0;
+        int replied = 0;
+        int lossSeconds = 0;
+        for (long second = finalSecond; second > finalSecond - seconds; second--)
+        {
+            int slot = Slot(second);
+            if (_second[slot] == second)
+            {
+                sent += _sent[slot];
+                replied += _replied[slot];
+                if (_sent[slot] > _replied[slot])
+                {
+                    lossSeconds++;
+                }
+            }
+        }
+
+        return (sent, replied, lossSeconds);
+    }
+
+    private void UpdateRttTrend(long nowUs)
+    {
+        long second = nowUs / UsPerSecond;
+        if (second == _lastRttSecond || double.IsNaN(_rttFast) || double.IsNaN(_rttSlow))
+        {
+            return;
+        }
+
+        _lastRttSecond = second;
+        bool high = _rttFast > _rttSlow * _options.RttFactor + _options.RttFloorMs;
+        _rttHighSeconds = high ? _rttHighSeconds + 1 : 0;
+    }
+
+    private static int Slot(long second) => (int)(((second % Buckets) + Buckets) % Buckets);
+}

backend/shared/MoShiro.Measurement/Core/Contracts.cs

@@ -186,6 +186,8 @@ public sealed class BlockSpec
 
     public long Int(string name) => Params.GetInt(name);
 
+    public double Number(string name) => Params.GetNumber(name);
+
     public bool Bool(string name) => Params.GetBool(name);
 
     /// <summary>参数或端口变化时调和器重启这个积木（不含流编号：Agent 的流编号每次重建都会重新分配）。</summary>
@@ -221,6 +223,9 @@ public sealed class BlockEnv
 
     public QuicDatagramTransport? Datagrams { get; init; }
 
+    /// <summary>拨号方：这条线路在每个时刻（单调时钟，微秒）丢包的历史概率（0 到 1），自适应探测用它在高发时段提前升频；没有历史为 null。</summary>
+    public Func<long, double?>? LossPrior { get; init; }
+
     /// <summary>监听方的准入策略（限速等）；拨号方为 null。</summary>
     public IAdmission? Admission { get; init; }
 
@@ -257,6 +262,7 @@ public sealed class BlockEnv
             Log = Log,
             Session = () => session,
             Datagrams = Datagrams,
+            LossPrior = LossPrior,
             Admission = Admission,
             Counters = Counters,
             TcpListen = TcpListen,

backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs

@@ -148,7 +148,7 @@ public sealed class BudgetGuard
 
     private static long EstimatePps(string kindId, ParamValues values) => kindId switch
     {
-        "udp_echo" or "tcp_echo" => values.GetInt("rate_hz") * 2,
+        "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kindId, values) * 2,
         "icmp_echo" => values.GetInt("rate_hz") * 2,
         "udp_train" => (long)(values.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, values.GetInt("size"))),
         "connect" => values.GetInt("per_minute") * 6 / 60,

backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs

@@ -8,10 +8,14 @@ public static class Kinds
 {
     private static readonly string[] s_directions = ["fwd", "rev", "bidir"];
 
+    /// <summary>这个任务可能达到的最高探测频率（Hz）：自适应时是 <c>rate_max_hz</c>，否则是固定的 <c>rate_hz</c>。预算、限速和准入都按它算。</summary>
+    public static long PeakRateHz(string kindId, ParamValues values) =>
+        (kindId is "udp_echo" or "tcp_echo") && values.GetBool("adaptive") ? Math.Max(values.GetInt("rate_max_hz"), values.GetInt("rate_min_hz")) : values.GetInt("rate_hz");
+
     private static readonly StreamSpec s_echoMsg = new(
         "msg",
         [
-            Col.Seq("seq"), Col.U16("size"),
+            Col.Seq("seq"), Col.U16("size"), Col.U16("w"),
             Col.TsDialer("t1"), Col.TsListener("t2", nullable: true), Col.TsListener("t3", nullable: true), Col.TsDialer("t4", nullable: true),
         ],
         Side.Dialer,
@@ -33,12 +37,17 @@ public static class Kinds
         Scope.Line,
         new BlockSet("quic_datagram", "fixed", "fixed", "echo", []),
         [
-            P.Int("rate_hz", 10, 1, 200, "Hz"),
+            P.Bool("adaptive", true, "Adapt the probe rate to the loss rate: low when the line is quiet, up to rate_max_hz when it loses packets, exponential back-off when it is unreachable."),
+            P.Int("rate_hz", 10, 1, 200, "Hz", "Fixed rate, used when adaptive is false."),
+            P.Int("rate_min_hz", 2, 1, 200, "Hz"),
+            P.Int("rate_max_hz", 10, 1, 200, "Hz", "Highest rate, and the reference grid for sample weights."),
+            P.Number("tau_pct", 0.2, 0.01, 5, "%", "Loss rate a quiet line has to be able to prove below, at 95% confidence."),
+            P.Number("rtt_tol_pct", 2, 0.1, 20, "%", "Half-width of the 95% confidence interval of the RTT p99 a quiet line has to keep, as a share of p99 (never below 0.3 ms)."),
             P.Int("size", 64, 44, 1400, "B"),
             P.Int("port", 7443, 1, 65535),
         ],
         [s_echoMsg],
-        p => Bandwidth.UdpEcho(p.GetInt("rate_hz"), p.GetInt("size")));
+        p => Bandwidth.UdpEcho(PeakRateHz("udp_echo", p), p.GetInt("size")));
 
     public static readonly KindSpec UdpTrain = new(
         "udp_train",
@@ -66,15 +75,20 @@ public static class Kinds
         Scope.Line,
         new BlockSet("tcp", "fixed", "uniform", "echo", ["tcp_info"]),
         [
-            P.Int("rate_hz", 20, 1, 200, "Hz"),
+            P.Bool("adaptive", true, "Adapt the echo rate: as low as the loss bound and the RTT percentile precision allow when the line is quiet, the maximum when it is not, back-off when it stalls."),
+            P.Int("rate_hz", 20, 1, 200, "Hz", "Fixed rate, used when adaptive is false."),
+            P.Int("rate_min_hz", 2, 1, 200, "Hz"),
+            P.Int("rate_max_hz", 20, 1, 200, "Hz", "Highest rate, and the reference grid for sample weights."),
+            P.Number("tau_pct", 0.2, 0.01, 5, "%", "Share of retransmitted segments a quiet line has to be able to prove below, at 95% confidence."),
+            P.Number("rtt_tol_pct", 2, 0.1, 20, "%", "Half-width of the 95% confidence interval of the RTT p99 a quiet line has to keep, as a share of p99 (never below 0.3 ms)."),
             P.Int("min", 32, 16, 1400, "B"),
             P.Int("max", 64, 16, 1400, "B"),
-            P.Int("tcp_info_ms", 100, 10, 1000, "ms"),
+            P.Int("tcp_info_ms", 1000, 10, 1000, "ms"),
             P.Int("port", 7444, 1, 65535),
             P.Bool("tls", false),
         ],
         [s_echoMsg, s_tcpInfo],
-        p => Bandwidth.TcpEcho(p.GetInt("rate_hz"), p.GetInt("max")));
+        p => Bandwidth.TcpEcho(PeakRateHz("tcp_echo", p), p.GetInt("max")));
 
     public static readonly KindSpec TcpBulk = new(
         "tcp_bulk",

backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs

@@ -19,6 +19,7 @@ public static class MoShiroMethodTable
                 MethodSpec.Request(MoShiroMethods.TestRun, RpcSide.Server),
                 MethodSpec.Request(MoShiroMethods.TestCancel, RpcSide.Server),
                 MethodSpec.Request(MoShiroMethods.DialerDisconnect, RpcSide.Server),
+                MethodSpec.Request(MoShiroMethods.LogsFetch, RpcSide.Server),
                 MethodSpec.Notification(MoShiroMethods.LinkRetry, RpcSide.Server),
                 MethodSpec.Notification(MoShiroMethods.UpdateOffer, RpcSide.Server),
                 MethodSpec.Notification(MoShiroMethods.UpdateApply, RpcSide.Server),

backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs

@@ -17,6 +17,7 @@ public static class MoShiroMethods
     public const string UpdateOffer = "moshiro.update.offer";
     public const string UpdateApply = "moshiro.update.apply";
     public const string Schedule = "moshiro.schedule";
+    public const string LogsFetch = "moshiro.logs.fetch";
 
     // 拨号方 ↔ 监听方（点对点会话）
     public const string Clock = "moshiro.clock";
@@ -338,11 +339,84 @@ public sealed class NodeHeartbeatExtra
     [JsonPropertyName("update")]
     public UpdateStatus? Update { get; set; }
 
-    [JsonPropertyName("runtime_logs")]
-    public List<NodeLogRecord>? RuntimeLogs { get; set; }
+    /// <summary>日志只报计数（不带文本）：自上次心跳以来各级别的条数、最多几个出现最多的类型，以及最新一条的编号；文本由 Backend 用 <c>moshiro.logs.fetch</c> 按需拉取。</summary>
+    [JsonPropertyName("log_summary")]
+    public LogSummaryDto? LogSummary { get; set; }
+}
 
-    [JsonPropertyName("dropped_runtime_logs")]
-    public long? DroppedRuntimeLogs { get; set; }
+/// <summary>某一类日志（消息文本的稳定哈希）在一个级别上出现了多少次。</summary>
+public sealed class LogTypeCountDto
+{
+    [JsonPropertyName("type")]
+    public uint Type { get; set; }
+
+    [JsonPropertyName("level")]
+    public string Level { get; set; } = "info";
+
+    [JsonPropertyName("count")]
+    public long Count { get; set; }
+}
+
+public sealed class LogSummaryDto
+{
+    [JsonPropertyName("last_id")]
+    public long LastId { get; set; }
+
+    [JsonPropertyName("info")]
+    public long Info { get; set; }
+
+    [JsonPropertyName("warn")]
+    public long Warn { get; set; }
+
+    [JsonPropertyName("error")]
+    public long Error { get; set; }
+
+    [JsonPropertyName("types")]
+    public List<LogTypeCountDto> Types { get; set; } = [];
+}
+
+/// <summary>Backend → 节点：拉取编号大于 <c>after_id</c> 的日志（最旧的在前），级别不低于 <c>min_level</c>。</summary>
+public sealed class LogsFetchParams
+{
+    [JsonPropertyName("after_id")]
+    public long AfterId { get; set; }
+
+    [JsonPropertyName("limit")]
+    public int Limit { get; set; } = 100;
+
+    [JsonPropertyName("min_level")]
+    public string? MinLevel { get; set; }
+}
+
+public sealed class LogEntryDto
+{
+    [JsonPropertyName("id")]
+    public long Id { get; set; }
+
+    [JsonPropertyName("type")]
+    public uint Type { get; set; }
+
+    [JsonPropertyName("at_utc_ms")]
+    public long AtUtcMs { get; set; }
+
+    [JsonPropertyName("level")]
+    public string Level { get; set; } = "info";
+
+    [JsonPropertyName("message")]
+    public string Message { get; set; } = "";
+
+    [JsonPropertyName("fields")]
+    public Dictionary<string, string> Fields { get; set; } = [];
+}
+
+public sealed class LogsFetchResult
+{
+    [JsonPropertyName("items")]
+    public List<LogEntryDto> Items { get; set; } = [];
+
+    /// <summary>节点上最新一条的编号：比 <c>after_id</c> 小说明节点重启过，编号从头开始。</summary>
+    [JsonPropertyName("last_id")]
+    public long LastId { get; set; }
 }
 
 /// <summary>
@@ -749,6 +823,9 @@ public static class FragmentKeys
 [JsonSerializable(typeof(DialTargets))]
 [JsonSerializable(typeof(UpdateOfferParams))]
 [JsonSerializable(typeof(ScheduleParams))]
+[JsonSerializable(typeof(LogsFetchParams))]
+[JsonSerializable(typeof(LogsFetchResult))]
+[JsonSerializable(typeof(LogSummaryDto))]
 [JsonSerializable(typeof(UpdateStatus))]
 [JsonSerializable(typeof(NodeHeartbeatExtra))]
 [JsonSerializable(typeof(NodeHelloMeta))]

backend/shared/tests/MoShiro.Measurement.Tests/AdaptiveRateTests.cs

@@ -0,0 +1,175 @@
+using MoShiro.Measurement;
+
+namespace MoShiro.Measurement.Tests;
+
+public sealed class AdaptiveRateTests
+{
+    private const long Second = 1_000_000;
+
+    /// <summary>按当前频率发包、按 <paramref name="lost"/> 决定回不回，每秒评估一次；返回结束时的秒数。</summary>
+    private static long Run(AdaptiveRate rate, long fromSecond, int seconds, Func<long, bool> lost, double rttMs = 20, List<ProbeTransition>? log = null, bool stopAtDead = false)
+    {
+        long second = fromSecond;
+        for (int i = 0; i < seconds; i++, second++)
+        {
+            int probes = Math.Max(1, (int)Math.Round(rate.Hz));
+            if (rate.State == ProbeState.Dead)
+            {
+                probes = i % 4 == 0 ? 1 : 0; // 退避：几秒才发一个
+            }
+
+            for (int k = 0; k < probes; k++)
+            {
+                long sent = second * Second + k * Second / probes;
+                rate.OnSend(sent);
+                if (!lost(second))
+                {
+                    rate.OnReply(sent, sent + (long)(rttMs * 1000), rttMs);
+                }
+            }
+
+            if (rate.Evaluate((second + 1) * Second) is { } change)
+            {
+                log?.Add(change);
+                if (stopAtDead && change.To == ProbeState.Dead)
+                {
+                    return second + 1;
+                }
+            }
+        }
+
+        return second;
+    }
+
+    [Fact]
+    public void QuietLine_AfterWarmUp_ProbesJustEnoughToProveTheLossBound()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions());
+        long at = Run(rate, 0, 600, _ => false);
+        Assert.Equal(ProbeState.Quiet, rate.State);
+        Assert.Equal(5, rate.Hz, 0.01); // 3 / (0.2% × 300 s)
+        Assert.True(at > 0);
+    }
+
+    [Fact]
+    public void SustainedLoss_RaisesToTheMaximum_AndComesBackOnlyAfterTheHoldTimes()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions());
+        long at = Run(rate, 0, 600, _ => false);
+        at = Run(rate, at, 60, second => second % 10 == 0); // 每 10 秒丢一整秒的包
+        Assert.Equal(ProbeState.Active, rate.State);
+        Assert.Equal(10, rate.Hz);
+
+        at = Run(rate, at, 800, _ => false); // 不到 15 分钟：还在 Active
+        Assert.Equal(ProbeState.Active, rate.State);
+        at = Run(rate, at, 400, _ => false); // 超过 15 分钟：降到 Watch
+        Assert.Equal(ProbeState.Watch, rate.State);
+        Run(rate, at, 1900, _ => false); // 最后一次丢包 30 分钟之后：回到 Quiet
+        Assert.Equal(ProbeState.Quiet, rate.State);
+    }
+
+    [Fact]
+    public void AnIsolatedLoss_OnlyRaisesToWatch()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions());
+        long at = Run(rate, 0, 600, _ => false);
+        var log = new List<ProbeTransition>();
+        Run(rate, at, 120, second => second == at + 10, log: log);
+        Assert.Equal(ProbeState.Watch, rate.State);
+        Assert.DoesNotContain(log, change => change.To == ProbeState.Active);
+    }
+
+    [Fact]
+    public void Unreachable_BacksOffExponentially_AndRecoversOnTheFirstReply()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions(), new Random(1));
+        long at = Run(rate, 0, 600, _ => false);
+        at = Run(rate, at, 60, _ => true, stopAtDead: true);
+        Assert.Equal(ProbeState.Dead, rate.State);
+
+        var delays = new List<double>();
+        long now = at * Second;
+        for (int i = 0; i < 8; i++)
+        {
+            double delay = rate.NextDelay().TotalSeconds;
+            delays.Add(delay);
+            now += (long)(delay * Second);
+            rate.OnSend(now);
+        }
+
+        Assert.InRange(delays[0], 0.8, 1.2); // 从 1 秒开始
+        Assert.True(delays[3] > delays[0] * 4, string.Join(",", delays)); // 翻倍增长（有抖动）
+        Assert.All(delays, delay => Assert.InRange(delay, 0.8, 36.0)); // 最长 30 秒，抖动 ±20%
+
+        rate.OnReply(now, now + 30_000, 30);
+        Assert.Equal(ProbeState.Watch, rate.Evaluate(now + 2 * Second)!.To);
+    }
+
+    [Fact]
+    public void QuietLineWithTightRtt_MayDropBelowTheGrid_ButAJitteryOneStaysAtTheMaximum()
+    {
+        var tight = new AdaptiveRate(new AdaptiveRateOptions());
+        Run(tight, 0, 1200, _ => false, rttMs: 20); // RTT 恒定：p99 的置信区间几乎为零，降到丢包下界 5 Hz
+        Assert.Equal(ProbeState.Quiet, tight.State);
+        Assert.Equal(5, tight.Hz, 0.01);
+
+        var jittery = new AdaptiveRate(new AdaptiveRateOptions());
+        var random = new Random(7);
+        long second = 0;
+        for (int i = 0; i < 1200; i++, second++)
+        {
+            int probes = Math.Max(1, (int)Math.Round(jittery.Hz));
+            for (int k = 0; k < probes; k++)
+            {
+                long sent = second * Second + k * Second / probes;
+                double rtt = 20 + random.NextDouble() * random.NextDouble() * 60; // 长尾 20 到 80 ms
+                jittery.OnSend(sent);
+                jittery.OnReply(sent, sent + (long)(rtt * 1000), rtt);
+            }
+
+            jittery.Evaluate((second + 1) * Second);
+        }
+
+        Assert.Equal(10, jittery.Hz, 0.01); // 分位数精度保护：抖动大，样本不能少
+    }
+
+    [Fact]
+    public void Weight_IsTheTimeAProbeStandsFor_InUnitsOfTheGrid()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions());
+        Run(rate, 0, 1200, _ => false);
+        Assert.Equal(5, rate.Hz, 0.01);
+        rate.NextDelay();
+        Assert.Equal(2, rate.Weight); // 5 Hz 的一个包代表 2 个 10 Hz 网格间隔
+    }
+
+    [Fact]
+    public void RttWellAboveTheBaseline_RaisesToWatch()
+    {
+        var rate = new AdaptiveRate(new AdaptiveRateOptions());
+        long at = Run(rate, 0, 900, _ => false, rttMs: 20);
+        Assert.Equal(ProbeState.Quiet, rate.State);
+        Run(rate, at, 30, _ => false, rttMs: 120);
+        Assert.Equal(ProbeState.Watch, rate.State);
+    }
+
+    [Fact]
+    public void Prior_WarmsUpBeforeTheEveningPeak_AndLetsHistoricallyQuietHoursUseTheLowestRate()
+    {
+        const long peakStart = 1_000 * 3600; // 秒
+        double? Prior(long us)
+        {
+            long second = us / Second;
+            return second >= peakStart && second < peakStart + 4 * 3600 ? 0.8 : 0.0;
+        }
+
+        var rate = new AdaptiveRate(new AdaptiveRateOptions { Prior = Prior });
+        long at = Run(rate, 100_000, 1200, _ => false); // 远离高峰，历史平稳：证明窗口拉长，2 Hz
+        Assert.Equal(ProbeState.Quiet, rate.State);
+        Assert.Equal(2, rate.Hz, 0.01);
+
+        Run(rate, peakStart - 1500, 120, _ => false); // 距离高峰不到半小时：提前升到 Active
+        Assert.Equal(ProbeState.Active, rate.State);
+        Assert.True(at > 0);
+    }
+}

backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs

@@ -65,7 +65,7 @@ public sealed class MeasurementTests
     {
         StreamSpec msg = Kinds.UdpEcho.FindStream("msg")!;
         var dialer = new RowStore(msg, Side.Dialer, 64);
-        Assert.Equal(["size", "t1", "t4"], msg.ColumnsOf(Side.Dialer).Select(c => c.Name));
+        Assert.Equal(["size", "w", "t1", "t4"], msg.ColumnsOf(Side.Dialer).Select(c => c.Name));
         Assert.Equal(["t2", "t3"], msg.ColumnsOf(Side.Listener).Select(c => c.Name));
 
         for (int i = 0; i < 4; i++)
@@ -78,7 +78,7 @@ public sealed class MeasurementTests
         dialer.Set(2, dialer.ColumnIndex("t4"), 5000);
         RawStream cut = dialer.Cut(1, 4, includeKey: true);
         Assert.Equal(3, cut.Count);
-        Assert.Equal(["seq", "size", "t1", "t4"], cut.Columns.Select(c => c.Name));
+        Assert.Equal(["seq", "size", "w", "t1", "t4"], cut.Columns.Select(c => c.Name));
         Assert.Equal([1L, 2L, 3L], cut.Find("seq")!.Values);
         Assert.Equal([false, true, false], cut.Find("t4")!.Valid!);
 
@@ -111,10 +111,10 @@ public sealed class MeasurementTests
 
     private static RawChunk EchoFragment(params (uint Seq, long T1, long? T4)[] rows)
     {
-        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
         foreach ((uint seq, long t1, long? t4) in rows)
         {
-            msg.Add(seq, 64, t1, t4);
+            msg.Add(seq, 64, 1, t1, t4);
         }
 
         return new RawChunk(new RawChunkHeader { Kind = "udp_echo", Clock = new ChunkClock { DialerOffsetUs = 1000 } }, [msg.Build()]);
@@ -131,7 +131,7 @@ public sealed class MeasurementTests
 
         Result<IReadOnlyList<RawStream>> merged = FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, source);
         RawStream msg = Assert.Single(merged.Value);
-        Assert.Equal(["seq", "size", "t1", "t2", "t3", "t4"], msg.Columns.Select(c => c.Name));
+        Assert.Equal(["seq", "size", "w", "t1", "t2", "t3", "t4"], msg.Columns.Select(c => c.Name));
         Assert.Equal([true, false, true], Enumerable.Range(0, 3).Select(i => msg.Find("t2")!.IsValid(i)));
         Assert.Equal(2050, msg.Find("t3")!.Values[0]);
         Assert.False(msg.Find("t4")!.IsValid(1));

backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs

@@ -92,6 +92,7 @@ public sealed class RawChunkCodecTests
     public void GoldenBody_LayoutIsStable()
     {
         // 布局变化（列编码、头部字段、字节序）会让这个测试失败：必须同时升级 container_version 并更新黄金字节。
+        // 只是回显流多了一列权重 w（块头自描述新增列，不用升级 container_version），黄金字节里多出六个 01。
         RawChunk chunk = SampleChunk();
         chunk.Header.Clock = new ChunkClock { ListenerMonoUs = 1000, UtcMs = 1_700_000_000_000, DialerOffsetUs = 500 };
         byte[] body = RawChunkCodec.EncodeBody(chunk);
@@ -104,7 +105,7 @@ public sealed class RawChunkCodecTests
         Assert.Equal(GoldenColumns, columns);
     }
 
-    private const string GoldenColumns = "00020202020240404040404080fbf3b407c09a0cc09a0cc09a0cc09a0cc09a0c3bc0e0d8b907889c0c90b818889c0c889c0c3ba4e1d8b907889c0c90b818889c0c889c0c2bb4fcf8b407889c0c90b81890b818";
+    private const string GoldenColumns = "00020202020240404040404001010101010180fbf3b407c09a0cc09a0cc09a0cc09a0cc09a0c3bc0e0d8b907889c0c90b818889c0c889c0c3ba4e1d8b907889c0c90b818889c0c889c0c2bb4fcf8b407889c0c90b81890b818";
 
     [Fact]
     public void Decode_RejectsBadMagicVersionCrcAndTruncation()

backend/src/MoShiro.Backend.Analysis/Engine/ChunkView.cs

@@ -33,5 +33,8 @@ public sealed class ChunkView
     public double ParamNumber(string name, double fallback) =>
         Header.Params.TryGetValue(name, out JsonElement v) && v.ValueKind == JsonValueKind.Number && v.TryGetDouble(out double d) ? d : fallback;
 
+    public bool ParamBool(string name, bool fallback) =>
+        Header.Params.TryGetValue(name, out JsonElement v) && v.ValueKind is JsonValueKind.True or JsonValueKind.False ? v.GetBoolean() : fallback;
+
     private static long FloorDiv(long a, long b) => (a / b) - ((a % b != 0 && (a < 0) != (b < 0)) ? 1 : 0);
 }

backend/src/MoShiro.Backend.Analysis/Ops/EchoOps.cs

@@ -26,6 +26,7 @@ internal readonly struct EchoColumns
         T2 = stream.Find("t2");
         T3 = stream.Find("t3");
         T4 = stream.Find("t4");
+        W = stream.Find("w");
     }
 
     public int Count { get; }
@@ -40,6 +41,11 @@ internal readonly struct EchoColumns
 
     public RawColumn? T4 { get; }
 
+    /// <summary>每个包的权重（它代表多少个基础网格间隔）；旧数据和固定频率的任务没有这一列，按 1 算。</summary>
+    public RawColumn? W { get; }
+
+    public int Weight(int i) => W is null || !W.IsValid(i) ? 1 : (int)Math.Clamp(W.Values[i], 1, 65535);
+
     public bool HasBasics => T1 is not null && T4 is not null;
 
     /// <summary>行 i 的发送时刻（UTC 毫秒）。</summary>
@@ -75,7 +81,7 @@ internal readonly struct EchoColumns
 
 internal sealed class RttOp : MetricOp
 {
-    public override string Signature => "rtt/1";
+    public override string Signature => "rtt/2";
 
     public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
     {
@@ -100,7 +106,11 @@ internal sealed class RttOp : MetricOp
 
             if (rttUs >= 0)
             {
-                sink.Emit(c.SentUtcMs(view, i), rttUs / 1000.0);
+                // 权重 k：这个样本代表 k 个网格间隔，按 k 次累计，分位数就是按时间平均的
+                for (int k = c.Weight(i); k > 0; k--)
+                {
+                    sink.Emit(c.SentUtcMs(view, i), rttUs / 1000.0);
+                }
             }
         }
     }
@@ -160,7 +170,7 @@ internal sealed class SlowShareOp : MetricOp
         double threshold = Math.Max(baseline * 1.8, baseline + Math.Max(15, baseline * 0.35));
         foreach ((int row, double ms) in rtts)
         {
-            sink.Emit(c.SentUtcMs(view, row), ms >= threshold ? 1 : 0, 1);
+            sink.Emit(c.SentUtcMs(view, row), ms >= threshold ? c.Weight(row) : 0, c.Weight(row));
         }
     }
 }
@@ -174,7 +184,7 @@ internal sealed class LossOp : MetricOp
         _dir = dir;
     }
 
-    public override string Signature => "loss/" + _dir + "/1";
+    public override string Signature => "loss/" + _dir + "/2";
 
     public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
     {
@@ -194,18 +204,18 @@ internal sealed class LossOp : MetricOp
             switch (_dir)
             {
                 case Dir.Up:
-                    sink.Emit(c.SentUtcMs(view, i), c.IsUpLost(i) ? 1 : 0, 1);
+                    sink.Emit(c.SentUtcMs(view, i), c.IsUpLost(i) ? c.Weight(i) : 0, c.Weight(i));
                     break;
                 case Dir.Down:
                     // 只统计上行已经到达 Agent 的包：回复没回来就是下行丢失。
                     if (!c.IsUpLost(i))
                     {
-                        sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? 1 : 0, 1);
+                        sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? c.Weight(i) : 0, c.Weight(i));
                     }
 
                     break;
                 default:
-                    sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? 1 : 0, 1);
+                    sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? c.Weight(i) : 0, c.Weight(i));
                     break;
             }
         }
@@ -452,7 +462,7 @@ internal sealed class GapOp : MetricOp
         _thresholdMs = thresholdMs;
     }
 
-    public override string Signature => $"gap/{_output}/{_rateParam}/{_thresholdMs}/1";
+    public override string Signature => $"gap/{_output}/{_rateParam}/{_thresholdMs}/2";
 
     public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
     {
@@ -462,11 +472,13 @@ internal sealed class GapOp : MetricOp
             return;
         }
 
-        double rate = Math.Max(0.1, view.ParamNumber(_rateParam, 1));
-        double expectedMs = 1000.0 / rate;
+        // 自适应频率时，每个包的权重就是它和上一个包的间隔（以 1/rate_max_hz 为单位），期望间隔按它算
+        bool adaptive = view.ParamBool("adaptive", false);
+        double rate = Math.Max(0.1, view.ParamNumber(adaptive ? "rate_max_hz" : _rateParam, 1));
         int[] order = c.ArrivalOrder();
         for (int i = 1; i < order.Length; i++)
         {
+            double expectedMs = 1000.0 * (adaptive ? c.Weight(order[i]) : 1) / rate;
             double gapMs = (c.T4!.Values[order[i]] - c.T4.Values[order[i - 1]]) / 1000.0;
             if (gapMs - expectedMs <= _thresholdMs)
             {

backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs

@@ -23,7 +23,7 @@ public static class AgentEndpoints
             .Api("agents_get", "One agent", AgentsTag).Returns<AgentDto>().ProducesProblems(401, 403, 404)
             .Requires(Capability.FleetRead);
         group.MapGet("/nodes/{id}/logs", ListNodeLogsAsync)
-            .Api("node_runtime_logs", "Recent runtime logs of a node", "admin-nodes").Returns<NodeRuntimeLogsDto>().ProducesProblems(401, 403)
+            .Api("node_runtime_logs", "Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept", "admin-nodes").Returns<NodeRuntimeLogsDto>().ProducesProblems(401, 403)
             .Requires(Capability.FleetRead);
         group.MapPatch("/agents/{id}", PatchAsync)
             .Accepts<AgentPatchRequest>("application/merge-patch+json", "application/json")
@@ -69,10 +69,13 @@ public static class AgentEndpoints
     private static async Task<IResult> ListAsync(AgentService agents, CancellationToken ct) =>
         TypedResults.Json(new PageDto<AgentDto>([.. (await agents.ListAsync(ct)).Select(Mappers.ToDto)], null), ApiJsonContext.Default.PageDtoAgentDto);
 
-    private static Microsoft.AspNetCore.Http.HttpResults.JsonHttpResult<NodeRuntimeLogsDto> ListNodeLogsAsync(NodeRuntimeLogService logs, string id, int? limit) =>
-        TypedResults.Json(
-            ToDto(logs.List(new NodeId(id), Math.Clamp(limit ?? 100, 1, 200))),
-            ApiJsonContext.Default.NodeRuntimeLogsDto);
+    /// <summary>先向节点拉一次最新的日志（离线或超时就算了），再返回 Backend 里保留的；页面开着时前端每几秒请求一次，就是实时拉取。</summary>
+    private static async Task<IResult> ListNodeLogsAsync(NodeRuntimeLogService logs, string id, int? limit, CancellationToken ct)
+    {
+        var node = new NodeId(id);
+        await logs.PullAsync(node, 100, null, ct);
+        return TypedResults.Json(ToDto(logs.List(node, Math.Clamp(limit ?? 100, 1, 200))), ApiJsonContext.Default.NodeRuntimeLogsDto);
+    }
 
     private static NodeRuntimeLogsDto ToDto(NodeRuntimeLogSnapshot snapshot) =>
         new([.. snapshot.Items.Select(item => new NodeRuntimeLogDto(item.Id, Mappers.Time(item.AtUtcMs), item.Level, item.Message, item.Fields))], snapshot.Dropped);

backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs

@@ -128,7 +128,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
                 return;
             }
 
-            _nodeLogs.Append(id, beat.RuntimeLogs, beat.DroppedRuntimeLogs ?? 0);
+            _nodeLogs.Observe(id, beat.LogSummary);
             state.ReplaceDialers(beat.Dialers.Select(d => new ConnectedDialerInfo(d.Id.Value, d.Version, d.Remote, d.SinceUtcMs, d.IpVersion)));
             foreach (ConnectedDialer dialer in beat.Dialers)
             {

backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs

@@ -8,47 +8,89 @@ public sealed record NodeRuntimeLog(long Id, long AtUtcMs, string Level, string
 
 public sealed record NodeRuntimeLogSnapshot(IReadOnlyList<NodeRuntimeLog> Items, long Dropped);
 
-/// <summary>Bounded in-memory recent logs uploaded by authenticated Agent and 拨号方 heartbeats.</summary>
-public sealed class NodeRuntimeLogService(TimeProvider time)
+/// <summary>
+/// 节点日志：心跳只带各级别的计数（不带文本），文本由这里按需向节点拉取。管理端打开日志页时拉最新的；
+/// 节点报告了错误，也会自动拉一次警告以上的日志留着，供事后查看（Backend 内存里每个节点保留最近 <see cref="CapacityPerNode"/> 条）。
+/// </summary>
+public sealed class NodeRuntimeLogService(TimeProvider time, IAgentGateway gateway)
 {
-    private const int CapacityPerNode = 256;
-    private const int MaxBatch = 32;
+    private const int CapacityPerNode = 512;
+    private const int PullLimit = 200;
+    private static readonly TimeSpan s_autoPullEvery = TimeSpan.FromMinutes(1);
+
     private readonly ConcurrentDictionary<NodeId, LogBuffer> _nodes = new();
 
-    public void Append(NodeId node, IReadOnlyList<NodeLogRecord>? records, long droppedByNode)
+    /// <summary>心跳里的日志计数：有新的错误就在后台自动拉一次警告以上的日志。</summary>
+    public void Observe(NodeId node, LogSummaryDto? summary)
     {
-        if ((records is null || records.Count == 0) && droppedByNode <= 0)
+        if (summary is null || summary.Error <= 0)
         {
             return;
         }
 
         LogBuffer buffer = _nodes.GetOrAdd(node, static _ => new LogBuffer());
+        long now = time.GetUtcNow().ToUnixTimeMilliseconds();
         lock (buffer.Gate)
         {
-            buffer.DroppedByNode += Math.Max(0, droppedByNode);
-            if (records is null)
+            if (buffer.Pulling || now - buffer.LastAutoPullMs < s_autoPullEvery.TotalMilliseconds)
             {
                 return;
             }
 
-            foreach (NodeLogRecord record in records.Take(MaxBatch))
+            buffer.LastAutoPullMs = now;
+        }
+
+        _ = Task.Run(async () =>
+        {
+            try
             {
-                long now = time.GetUtcNow().ToUnixTimeMilliseconds();
-                long at = record.AtUtcMs < now - 86_400_000 || record.AtUtcMs > now + 86_400_000 ? now : record.AtUtcMs;
-                var entry = new NodeRuntimeLog(
-                    ++buffer.LastId,
-                    at,
-                    record.Level is "info" or "warn" or "error" ? record.Level : "info",
-                    Truncate(record.Message ?? "", 384),
-                    SafeFields(record.Fields));
+                await PullAsync(node, PullLimit, "warn", CancellationToken.None);
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException)
+            {
+                // 拉取是尽力而为：节点刚好断开或超时，下次有新错误再拉
+            }
+        });
+    }
 
-                if (buffer.Items.Count == CapacityPerNode)
-                {
-                    buffer.Items.Dequeue();
-                    buffer.DroppedByBackend++;
-                }
+    /// <summary>向节点拉取上次之后的新日志；节点离线或超时返回 false，已有的日志照常可看。</summary>
+    public async ValueTask<bool> PullAsync(NodeId node, int limit, string? minLevel, CancellationToken ct)
+    {
+        LogBuffer buffer = _nodes.GetOrAdd(node, static _ => new LogBuffer());
+        long after;
+        lock (buffer.Gate)
+        {
+            if (buffer.Pulling)
+            {
+                return true; // 已经有一次拉取在进行，它的结果同样会进缓冲区
+            }
 
-                buffer.Items.Enqueue(entry);
+            buffer.Pulling = true;
+            after = buffer.LastFetchedId;
+        }
+
+        try
+        {
+            using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
+            timeout.CancelAfter(TimeSpan.FromSeconds(5));
+            Result<LogsFetchResult> result = await gateway.FetchLogsAsync(node, new LogsFetchParams { AfterId = after, Limit = Math.Clamp(limit, 1, PullLimit), MinLevel = minLevel }, timeout.Token);
+            if (!result.TryGet(out LogsFetchResult? page, out _))
+            {
+                return false;
+            }
+
+            Append(buffer, after, page);
+            return true;
+        }
+        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
+        {
+            return false;
+        }
+        finally
+        {
+            lock (buffer.Gate)
+            {
+                buffer.Pulling = false;
             }
         }
     }
@@ -62,9 +104,56 @@ public sealed class NodeRuntimeLogService(TimeProvider time)
 
         lock (buffer.Gate)
         {
-            return new NodeRuntimeLogSnapshot(
-                [.. buffer.Items.Reverse().Take(Math.Clamp(limit, 1, 200))],
-                buffer.DroppedByNode + buffer.DroppedByBackend);
+            return new NodeRuntimeLogSnapshot([.. buffer.Items.Reverse().Take(Math.Clamp(limit, 1, 200))], buffer.Dropped);
+        }
+    }
+
+    private void Append(LogBuffer buffer, long after, LogsFetchResult page)
+    {
+        long now = time.GetUtcNow().ToUnixTimeMilliseconds();
+        lock (buffer.Gate)
+        {
+            if (page.LastId < after)
+            {
+                // 节点重启过，编号从头开始：旧的编号不再有意义
+                buffer.LastFetchedId = 0;
+                after = 0;
+            }
+
+            long expected = after + 1;
+            foreach (LogEntryDto entry in page.Items)
+            {
+                if (entry.Id <= buffer.LastFetchedId)
+                {
+                    continue;
+                }
+
+                if (entry.Id > expected && buffer.LastFetchedId > 0)
+                {
+                    buffer.Dropped += entry.Id - expected; // 中间的几条在节点上已经被环挤掉了
+                }
+
+                expected = entry.Id + 1;
+                long at = entry.AtUtcMs < now - 86_400_000 || entry.AtUtcMs > now + 86_400_000 ? now : entry.AtUtcMs;
+                if (buffer.Items.Count == CapacityPerNode)
+                {
+                    buffer.Items.Dequeue();
+                    buffer.Dropped++;
+                }
+
+                buffer.Items.Enqueue(new NodeRuntimeLog(
+                    entry.Id,
+                    at,
+                    entry.Level is "info" or "warn" or "error" ? entry.Level : "info",
+                    Truncate(entry.Message, 384),
+                    SafeFields(entry.Fields)));
+                buffer.LastFetchedId = entry.Id;
+            }
+
+            if (page.Items.Count == 0 && page.LastId > buffer.LastFetchedId)
+            {
+                buffer.LastFetchedId = page.LastId; // 没有符合级别的新日志：下次从这里接着拉，不用重复扫
+            }
         }
     }
 
@@ -80,17 +169,11 @@ public sealed class NodeRuntimeLogService(TimeProvider time)
 
         foreach ((string key, string value) in fields.Take(20))
         {
-            string safeKey = Truncate(key, 48);
             string lower = key.ToLowerInvariant();
-            bool sensitive = lower.Contains("token", StringComparison.Ordinal)
-                || lower.Contains("secret", StringComparison.Ordinal)
-                || lower.Contains("password", StringComparison.Ordinal)
-                || lower.Contains("private", StringComparison.Ordinal)
-                || lower.Contains("signature", StringComparison.Ordinal)
-                || lower.Contains("authorization", StringComparison.Ordinal)
-                || lower.Contains("cookie", StringComparison.Ordinal)
-                || lower == "mac";
-            safe[safeKey] = sensitive ? "[redacted]" : Truncate(value ?? "", 128);
+            bool sensitive = lower.Contains("token", StringComparison.Ordinal) || lower.Contains("secret", StringComparison.Ordinal) || lower.Contains("password", StringComparison.Ordinal)
+                || lower.Contains("private", StringComparison.Ordinal) || lower.Contains("signature", StringComparison.Ordinal) || lower.Contains("authorization", StringComparison.Ordinal)
+                || lower.Contains("cookie", StringComparison.Ordinal) || lower == "mac";
+            safe[Truncate(key, 64)] = sensitive ? "[redacted]" : Truncate(value, 128);
         }
 
         return safe;
@@ -99,9 +182,15 @@ public sealed class NodeRuntimeLogService(TimeProvider time)
     private sealed class LogBuffer
     {
         public object Gate { get; } = new();
-        public Queue<NodeRuntimeLog> Items { get; } = [];
-        public long LastId;
-        public long DroppedByNode;
-        public long DroppedByBackend;
+
+        public Queue<NodeRuntimeLog> Items { get; } = new();
+
+        public long LastFetchedId { get; set; }
+
+        public long Dropped { get; set; }
+
+        public long LastAutoPullMs { get; set; }
+
+        public bool Pulling { get; set; }
     }
 }

backend/src/MoShiro.Backend.Application/BackendOptions.cs

@@ -60,6 +60,9 @@ public sealed class BackendOptions
 
     public int LoginAttemptsPerMinute { get; set; } = 10;
 
+    /// <summary>Agent 至少要是哪个版本才能连上（如 <c>0.4.2-dev.80</c>）；空 = 不要求。低于它的 Agent 握手时被拒绝，5 秒后重试（它可能正在更新）。环境变量 <c>MoShiro__MinAgentVersion</c>。</summary>
+    public string? MinAgentVersion { get; set; }
+
     public int RawRetentionDays { get; set; } = 30;
 
     public int TestRawRetentionDays { get; set; } = 90;

backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs

@@ -25,6 +25,9 @@ public interface IAgentGateway
 
     ValueTask CancelTestAsync(NodeId agent, TestId test, CancellationToken ct);
 
+    /// <summary>向节点拉取日志（编号大于 <c>AfterId</c> 的、不低于 <c>MinLevel</c> 的，最旧的在前）。节点离线时失败。</summary>
+    ValueTask<Result<LogsFetchResult>> FetchLogsAsync(NodeId node, LogsFetchParams request, CancellationToken ct);
+
     /// <summary>让监听方断开某个拨号方的会话。返回监听方是否真的有这个会话。</summary>
     ValueTask<Result<bool>> DisconnectDialerAsync(NodeId listener, NodeId dialer, CancellationToken ct);
 

backend/src/MoShiro.Backend.Gateway/FleetGateway.cs

@@ -50,6 +50,15 @@ public sealed class FleetGateway : IAgentGateway
         await Connections.RequestAsync(
             agent, MoShiroMethods.TestCancel, new TestCancelParams { TestId = test }, MoShiroJsonContext.Default.TestCancelParams, MoShiroJsonContext.Default.TestRunResult, ct);
 
+    public async ValueTask<Result<LogsFetchResult>> FetchLogsAsync(NodeId node, LogsFetchParams request, CancellationToken ct)
+    {
+        Result<LogsFetchResult> result = await Connections.RequestAsync(
+            node, MoShiroMethods.LogsFetch, request, MoShiroJsonContext.Default.LogsFetchParams, MoShiroJsonContext.Default.LogsFetchResult, ct);
+        return result.TryGet(out LogsFetchResult? page, out FleetError? error)
+            ? Result<LogsFetchResult>.Ok(page)
+            : Result<LogsFetchResult>.Fail(error.Code == FleetErrors.NodeOffline ? MoShiroErrors.AgentOffline : error.Code, error.Message);
+    }
+
     public async ValueTask<Result<bool>> DisconnectDialerAsync(NodeId listener, NodeId dialer, CancellationToken ct)
     {
         Result<TestRunResult> result = await Connections.RequestAsync(

backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs

@@ -16,6 +16,7 @@ public static class GatewayServiceCollectionExtensions
     public static IServiceCollection AddMoShiroGateway(this IServiceCollection services)
     {
         services.AddSingleton<IAgentGateway, FleetGateway>();
+        services.AddSingleton<IFleetVersionPolicy, BackendVersionPolicy>();
         services.AddHostedService<ConnectionShutdown>();
         services.AddFleetEnrollment();
         services.AddFleetServer(options => options.NodeIdPrefix = AgentIds.Prefix)
@@ -44,3 +45,11 @@ public static class GatewayServiceCollectionExtensions
                 (context, parameters, ct) => services.GetRequiredService<TestService>().OnStatusAsync(context.Node.Id, parameters, ct));
     }
 }
+
+/// <summary>握手时的版本要求：告诉 Agent 面板的版本，并要求 Agent 不低于配置的最低版本。</summary>
+internal sealed class BackendVersionPolicy(BackendOptions options) : IFleetVersionPolicy
+{
+    public string? ServerVersion => BuildInfo.Version;
+
+    public string? MinNodeVersion => string.IsNullOrWhiteSpace(options.MinAgentVersion) ? null : options.MinAgentVersion;
+}

backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs

@@ -31,6 +31,25 @@ public sealed class OpTests
         Assert.InRange(rtt.Quantile(0.5)!.Value, 50 * 0.955, 50 * 1.045);
     }
 
+    [Fact]
+    public void SampleWeights_KeepRttQuantilesAndLossTimeAveraged_WhenTheProbeRateChanges()
+    {
+        static EchoPacket At(int i) => new(LostUp: i % 20 < 2, LostDown: false, i % 10 < 2 ? 100_000 : 20_000, 20_000);
+
+        // 每 100 ms 一个包，每个权重 1；对照：只取偶数行，每 200 ms 一个包，每个权重 2——代表同样长的时间
+        RawChunk dense = ChunkFactory.Echo(s_agent, "udp_echo", T0, 100, 100, At);
+        RawChunk sparse = ChunkFactory.Echo(s_agent, "udp_echo", T0, 50, 200, i => At(i * 2), weight: _ => 2);
+
+        MetricSet d = EvaluateOneBucket(dense, "rtt", "loss.up", "loss.total");
+        MetricSet s = EvaluateOneBucket(sparse, "rtt", "loss.up", "loss.total");
+        var dr = (HistogramAcc)d.Items["rtt"];
+        var sr = (HistogramAcc)s.Items["rtt"];
+        Assert.Equal(dr.Count, sr.Count); // 稀疏的 45 个回复各按权重 2 累计，和密集的 90 个一样
+        Assert.Equal(dr.Quantile(0.5), sr.Quantile(0.5));
+        Assert.Equal(dr.Quantile(0.95), sr.Quantile(0.95));
+        Assert.Equal(((RatioAcc)d.Items["loss.up"]).Percent, ((RatioAcc)s.Items["loss.up"]).Percent);
+    }
+
     [Fact]
     public void Loss_SeparatesUpAndDown()
     {

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

@@ -42,7 +42,7 @@ public sealed class AgentHarness : IAsyncDisposable
     public byte[] TlsPublicKey => _identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo();
 
     /// <param name="pin">带信任根固定的加入码（码、创世指纹、属主）：节点按固定了创世和属主的方式注册。</param>
-    public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null, (string Code, string Genesis, string[] Controllers)? pin = null)
+    public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null, (string Code, string Genesis, string[] Controllers)? pin = null, bool waitForConnection = true)
     {
         string directory = Path.Combine(Path.GetTempPath(), "moshiro-agent-" + Guid.NewGuid().ToString("N"));
         string token = pin?.Code ?? await backend.CreateJoinTokenAsync();
@@ -71,7 +71,11 @@ public sealed class AgentHarness : IAsyncDisposable
         }
 
         await app.Ready;
-        await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend");
+        if (waitForConnection)
+        {
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend");
+        }
+
         return harness;
     }
 

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

@@ -1,27 +1,53 @@
-using Fleet.Core;
-using MoShiro.Backend.Application;
+using System.Text.Json.Nodes;
 using MoShiro.Protocol;
+using MoShiro.TestSupport;
 
 namespace MoShiro.Backend.Tests;
 
-public sealed class NodeRuntimeLogTests
+public sealed class NodeRuntimeLogTests : IClassFixture<BackendFixture>
 {
-    [Fact]
-    public void DiagnosticFieldsSurviveUpload_WhileSecretsAndOversizedFieldsRemainBounded()
+    private readonly BackendFixture _backend;
+
+    public NodeRuntimeLogTests(BackendFixture backend)
     {
-        var service = new NodeRuntimeLogService(TimeProvider.System);
-        var node = new NodeId("agt_diagnostics");
+        _backend = backend;
+    }
+
+    private static LogEntryDto Entry(long id, string level, string message, Dictionary<string, string>? fields = null) =>
+        new() { Id = id, Type = (uint)id, AtUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), Level = level, Message = message, Fields = fields ?? [] };
+
+    private async Task<JsonNode> LogsAsync(AgentSimulator agent) =>
+        await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/nodes/{agent.NodeId.Value}/logs?limit=50");
+
+    [Fact]
+    public async Task Logs_ArePulledFromTheNodeOnDemand_OnlyTheNewLinesEachTime_WithSecretsAndOversizedFieldsBounded()
+    {
+        await using AgentSimulator agent = await _backend.ConnectAgentAsync();
         Dictionary<string, string> fields = Enumerable.Range(0, 18).ToDictionary(i => "field_" + i, _ => "value");
         fields.Add("api_token", "must-not-survive");
         fields.Add("detail", new string('x', 500));
         fields.Add("overflow", "discard");
-        service.Append(node, [new NodeLogRecord(TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), "warn", "handshake failed", fields)], 0);
+        agent.Logs.Add(Entry(1, "info", "link up"));
+        agent.Logs.Add(Entry(2, "warn", "handshake failed", fields));
 
-        NodeRuntimeLog entry = Assert.Single(service.List(node, 10).Items);
-        Assert.Equal(20, entry.Fields.Count);
-        Assert.Equal("value", entry.Fields["field_17"]);
-        Assert.Equal("[redacted]", entry.Fields["api_token"]);
-        Assert.Equal(128, entry.Fields["detail"].Length);
-        Assert.False(entry.Fields.ContainsKey("overflow"));
+        JsonNode first = await LogsAsync(agent);
+        Assert.Equal(["handshake failed", "link up"], first["items"]!.AsArray().Select(i => i!["message"]!.GetValue<string>())); // 新的在前
+        JsonNode warn = first["items"]![0]!;
+        Assert.Equal(20, warn["fields"]!.AsObject().Count);
+        Assert.Equal("[redacted]", warn["fields"]!["api_token"]!.GetValue<string>());
+        Assert.Equal(128, warn["fields"]!["detail"]!.GetValue<string>().Length);
+        Assert.Null(warn["fields"]!["overflow"]);
+
+        agent.Logs.Add(Entry(3, "error", "transport shutdown"));
+        JsonNode second = await LogsAsync(agent);
+        Assert.Equal(3, second["items"]!.AsArray().Count);
+        Assert.Equal(2, agent.LogFetches.Last().AfterId); // 第二次只拉编号 2 之后的
+    }
+
+    [Fact]
+    public async Task Logs_OfAnOfflineNode_AreWhatTheBackendKept()
+    {
+        JsonNode logs = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/nodes/agt_nobody/logs");
+        Assert.Empty(logs["items"]!.AsArray());
     }
 }

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

@@ -0,0 +1,72 @@
+using System.Text.Json;
+using Fleet.Core;
+using Fleet.Rpc;
+using Microsoft.Extensions.DependencyInjection;
+using MoShiro.Backend.Application;
+using MoShiro.TestSupport;
+
+namespace MoShiro.Backend.Tests;
+
+public sealed class VersionRequirementTests : IClassFixture<BackendFixture>
+{
+    private readonly BackendFixture _backend;
+
+    public VersionRequirementTests(BackendFixture backend)
+    {
+        _backend = backend;
+    }
+
+    [Fact]
+    public async Task AnAgentBelowTheRequiredVersion_IsRejectedAtHello_AndGetsInOnceTheRequirementIsMet()
+    {
+        BackendOptions options = _backend.App.Services.GetRequiredService<BackendOptions>();
+        options.MinAgentVersion = "9.0.0"; // 模拟器的版本是 0.0.1-sim
+        try
+        {
+            await using AgentSimulator old = await _backend.ConnectAgentAsync(hello: false);
+            var hello = new HelloParams { NodeVersion = "0.0.1-sim", NowUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() };
+            RpcMessage reply = await old.RequestRawAsync(FleetMethods.Hello, JsonSerializer.SerializeToElement(hello, FleetRpcContext.Default.HelloParams));
+            Assert.Equal(FleetErrors.VersionRequired.Value, reply.Error!.Data!.Code);
+            Assert.Contains("9.0.0", reply.Error.Message);
+        }
+        finally
+        {
+            options.MinAgentVersion = null; // 共用的后端：恢复，别影响别的测试
+        }
+
+        await using AgentSimulator ok = await _backend.ConnectAgentAsync();
+        Assert.True(_backend.App.Services.GetRequiredService<MoShiro.Backend.Application.IAgentGateway>().IsOnline(ok.NodeId));
+    }
+
+    [Fact]
+    [Trait("Category", "Slow")]
+    public async Task AnAgentTheBackendRejectedForItsVersion_TriesAgainEveryFiveSeconds_AndComesOnlineWhenAccepted()
+    {
+        BackendOptions options = _backend.App.Services.GetRequiredService<BackendOptions>();
+        IAgentGateway gateway = _backend.App.Services.GetRequiredService<IAgentGateway>();
+        options.MinAgentVersion = "999.0.0";
+        try
+        {
+            await using AgentHarness agent = await AgentHarness.StartAsync(_backend, waitForConnection: false);
+            await Task.Delay(TimeSpan.FromSeconds(2));
+            Assert.False(gateway.IsOnline(agent.NodeId));
+
+            options.MinAgentVersion = null; // 通过了：下一次握手（最迟 5 秒）就上线
+            await BackendFixture.EventuallyAsync(() => Task.FromResult(gateway.IsOnline(agent.NodeId)), TimeSpan.FromSeconds(12));
+        }
+        finally
+        {
+            options.MinAgentVersion = null;
+        }
+    }
+
+    [Fact]
+    [Trait("Category", "Slow")]
+    public async Task AnAgentThatRequiresANewerBackend_DoesNotGoOnline_AndKeepsChecking()
+    {
+        IAgentGateway gateway = _backend.App.Services.GetRequiredService<IAgentGateway>();
+        await using AgentHarness agent = await AgentHarness.StartAsync(_backend, config => config.Fleet.MinServerVersion = "999.0.0", waitForConnection: false);
+        await Task.Delay(TimeSpan.FromSeconds(7)); // 至少重试过一次
+        Assert.False(gateway.IsOnline(agent.NodeId));
+    }
+}

backend/tests/MoShiro.TestSupport/AgentSimulator.cs

@@ -17,6 +17,11 @@ namespace MoShiro.TestSupport;
 /// </summary>
 public sealed class AgentSimulator : IAsyncDisposable
 {
+    /// <summary>节点上保留的日志（<c>moshiro.logs.fetch</c> 按编号返回它们）和收到的拉取请求。</summary>
+    public System.Collections.Concurrent.ConcurrentBag<LogEntryDto> Logs { get; } = [];
+
+    public System.Collections.Concurrent.ConcurrentQueue<LogsFetchParams> LogFetches { get; } = new();
+
     private readonly NodeIdentity _identity;
     private readonly NodeIdentity _tlsKey;
     private readonly ClientWebSocket _socket = new();
@@ -446,6 +451,13 @@ public sealed class AgentSimulator : IAsyncDisposable
                     await SendAsync(RpcMessage.Failure(message.Id, new FleetError(new ErrorCode(errorCode ?? "test_busy"), "refused")));
                 }
 
+                break;
+            case MoShiroMethods.LogsFetch:
+                LogsFetchParams fetch = message.Params!.Value.Deserialize(MoShiroJsonContext.Default.LogsFetchParams)!;
+                LogFetches.Enqueue(fetch);
+                var page = new LogsFetchResult { LastId = Logs.IsEmpty ? 0 : Logs.Max(l => l.Id) };
+                page.Items.AddRange(Logs.Where(l => l.Id > fetch.AfterId).OrderBy(l => l.Id).Take(Math.Max(1, fetch.Limit)));
+                await SendAsync(RpcMessage.Success(message.Id!.Value, page, MoShiroJsonContext.Default.LogsFetchResult));
                 break;
             case MoShiroMethods.TestCancel:
                 TestCancelParams cancel = message.Params!.Value.Deserialize(MoShiroJsonContext.Default.TestCancelParams)!;

backend/tests/MoShiro.TestSupport/ChunkFactory.cs

@@ -48,7 +48,8 @@ public static class ChunkFactory
         Func<int, EchoPacket> packet,
         int durationMs = 10_000,
         IReadOnlyDictionary<string, JsonElement>? parameters = null,
-        IEnumerable<RawStream>? extraStreams = null)
+        IEnumerable<RawStream>? extraStreams = null,
+        Func<int, int>? weight = null)
     {
         var msg = new StreamBuilder(Kinds.Find(kind)!.FindStream("msg")!);
         for (int i = 0; i < count; i++)
@@ -58,7 +59,7 @@ public static class ChunkFactory
             long t2 = t1 + DialerOffsetUs + p.OwdUpUs;
             long t3 = t2 + p.ProcessUs;
             long t4 = t3 - DialerOffsetUs + p.OwdDownUs;
-            msg.Add(i, 64, t1, p.LostUp ? null : t2, p.LostUp ? null : t3, p.LostUp || p.LostDown ? null : t4);
+            msg.Add(i, 64, weight?.Invoke(i) ?? 1, t1, p.LostUp ? null : t2, p.LostUp ? null : t3, p.LostUp || p.LostDown ? null : t4);
         }
 
         var streams = new List<RawStream> { msg.Build() };

backend/tests/MoShiro.TestSupport/DialerSimulator.cs

@@ -160,10 +160,10 @@ public sealed class DialerSimulator : IAsyncDisposable
         _echoes.TryRemove(task.StreamId, out _);
         _quic.Unregister(task.StreamId);
 
-        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
         foreach ((uint seq, long t1) in collector.Sent.OrderBy(p => p.Key))
         {
-            msg.Add(seq, size, t1, collector.Received.TryGetValue(seq, out long t4) ? t4 : null);
+            msg.Add(seq, size, 1, t1, collector.Received.TryGetValue(seq, out long t4) ? t4 : null);
         }
 
         RawChunkHeader header = Header("udp_echo", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params);
@@ -195,7 +195,7 @@ public sealed class DialerSimulator : IAsyncDisposable
         await stream.WriteAsync(auth, ct);
 
         long startUtc = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
-        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+        var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
         int received = 0;
         for (uint seq = 0; seq < count; seq++)
         {
@@ -207,7 +207,7 @@ public sealed class DialerSimulator : IAsyncDisposable
             byte[]? reply = await TcpWire.ReadMessageAsync(stream, ct);
             long? t4 = reply is not null && TcpWire.TryReadEchoBody(reply, out uint echoed, out _) && echoed == seq ? NowUs() : null;
             received += t4 is null ? 0 : 1;
-            msg.Add(seq, body.Length, t1, t4);
+            msg.Add(seq, body.Length, 1, t1, t4);
             await Task.Delay(interval, ct);
         }
 

docs/design/protocol.md

@@ -17,7 +17,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上：
 | 更新 | 通知 `moshiro.update.offer` / `moshiro.update.apply`（B→A、B→P），心跳附加字段 `update` | `offer` 带 CI 签名的发布清单原文和签名，节点只写进数据目录；`apply` 请求更新；`update` 是宿主机更新器的状态（`state`、`current`、`available`、`error`、`at`），见 [deployment.md](../ops/deployment.md) |
 | 上报原始块 | 发件箱，二进制通道 16 | `RawChunk`（自描述，见 [measurement.md](measurement.md) 第 4 节），`msg_id = chunk_id` |
 | 拨号方 交付列块 | 点对点会话的发件箱，二进制通道 17 | 列块：`u32 stream_id \| u32 first_seq \| u32 count \| 列数据(Brotli)` |
-| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent：`sessions`、`dialers[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`；拨号方：只有 `update` |
+| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent：`sessions`、`dialers[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`、`log_summary`（自上次心跳以来各级别日志的条数、出现最多的几个类型 `type`（消息文本的稳定哈希）和最新一条的编号，不带文本）；拨号方：只有 `update` |
 | 版本与能力 | `fleet.hello` | capabilities 包括 `moshiro.kinds:<列表>`、`moshiro.blocks:<列表>`；Agent 附加 `platform`（系统、架构、版本、tag、提交、构建时间）、`tls_pubkey`（面向 拨号方 的证书公钥，面板自己算指纹）、`addresses`（本机公网地址）、`listen`、`local_limits` 摘要；拨号方 附加 `platform` |
 
 ## 2. `moshiro.*` 方法
@@ -32,6 +32,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上：
 | `moshiro.test.run` | B→A | 请求 | `test_id`、`dialer_id`（在哪个 拨号方 上跑）、`plan`、`phases[]` → `accepted`，或错误 `test_busy`、`dialer_offline` |
 | `moshiro.dialer.disconnect` | B→A | 请求 | `dialer_id` → `accepted`（Agent 是否真的有这个会话）；Agent 关闭该 拨号方 的会话 |
 | `moshiro.test.cancel` | B→A | 请求 | `test_id` |
+| `moshiro.logs.fetch` | B→A | 请求 | `after_id`、`limit`（≤ 200）、`min_level` → `items[]`（`id`、`type`、`at_utc_ms`、`level`、`message`、`fields`）、`last_id`：按需拉取节点本地保留的日志（最近 2048 条），最旧的在前；`last_id` 比 `after_id` 小说明节点重启过。管理端打开日志页时拉，节点报告了错误时 Backend 自动拉一次警告以上的 |
 | `moshiro.test.status` | A→B | 通知 | `test_id`、`status`（`running`、`phase_done`、`succeeded`、`failed`、`aborted`）、`phase?`、`error_code?` |
 | `moshiro.event` | A→B | 通知 | `kind`（`dialer_connected`（`detail` 是 `dialer_id`，`dialer` 里带版本、远端地址和连接时间）、`dialer_disconnected`、`endpoint_changed`、`session_rejected`、`clock_step`、`gap`、`cpu_saturated`（节点进程 CPU 饱和，可能影响测量，由 拨号方 或 Agent 上报））、`at_utc`、`detail` |
 | `moshiro.host` | A、P→B | 通知 | `HostSnapshot`：主机资源占用的汇总数字（CPU、内存、根分区、网卡累计流量与网速、运行时长；没有型号、主机名、IP），每 `host_interval_s` 秒一次；`host_stats` 为 false 时不发，见 [enrollment.md](../security/enrollment.md) |

web/shared/api/schema.ts

@@ -508,7 +508,7 @@ export interface paths {
             path?: never;
             cookie?: never;
         };
-        /** Recent runtime logs of a node */
+        /** Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept */
         get: operations["node_runtime_logs"];
         put?: never;
         post?: never;