feat(ops): upload node logs and task traffic [CI]

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

@@ -24,6 +24,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 MonoClock _clock;
     private readonly AgentCounters _counters = new();
     private readonly SessionRegistry _sessions = new();
@@ -37,6 +38,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
     private readonly TcpAcceptor _tcp;
     private readonly PeerListener _peers;
     private readonly PeerHub _hub;
+    private readonly ChunkAssembler _chunks;
     private readonly TaskManager _tasks;
     private readonly TestRunner _tests;
     private readonly FleetLink _link;
@@ -52,6 +54,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
         _identity = identity;
         _time = time;
         _log = log;
+        _logBatchSource = log as INodeLogBatchSource;
         _clock = new MonoClock(time);
         _uplink = new BackendUplink(time);
 
@@ -82,10 +85,10 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
             MaxPacketsPerSecond = config.Limits.MaxTotalPps,
             MaxBulkMbps = config.Limits.MaxTotalMbps,
         };
-        var chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
-        var targets = new TargetRunners(chunks, identity.NodeId, _clock, time, log);
+        _chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
+        var targets = new TargetRunners(_chunks, identity.NodeId, _clock, time, log);
         _tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _datagrams.BoundPorts, () => _tcp.BoundPorts, _trust);
-        _hub = new PeerHub(_sessions, _clock, chunks, _tasks, _uplink, time, log);
+        _hub = new PeerHub(_sessions, _clock, _chunks, _tasks, _uplink, time, log);
         _tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "quic_datagram" ? _datagrams.BoundPorts : _tcp.BoundPorts, time, log);
 
         _tasks.OnLineTasksChanged = (probe, set) => _ = _hub.PushToAsync(probe, set).AsTask();
@@ -179,11 +182,12 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
         _log.Info("agent listening", ("control", ControlPort), ("quic", string.Join(',', QuicPorts)), ("tcp", string.Join(',', TcpPorts)));
         try
         {
-            await Task.WhenAll(quic, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token));
+            await Task.WhenAll(quic, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token), _chunks.RunTrafficReportingAsync(linked.Token));
         }
         finally
         {
             await _tasks.StopAsync();
+            _chunks.FlushTraffic();
             await linked.CancelAsync();
         }
     }
@@ -351,6 +355,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);
         var extra = new AgentHeartbeatExtra
         {
             RuntimeStatus = _update?.Status().State == "applying" ? NodeRuntimeStatus.Updating : NodeRuntimeStatus.Running,
@@ -363,6 +368,8 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
             DroppedRateLimited = _counters.DroppedRateLimited,
             QueueBytes = _outbox.PendingBytes,
             Update = _update?.Status(),
+            RuntimeLogs = logBatch.Records.Count == 0 ? null : [.. logBatch.Records],
+            DroppedRuntimeLogs = logBatch.Dropped == 0 ? null : logBatch.Dropped,
         };
         _lastIn = packetsIn;
         _lastOut = packetsOut;

backend/agent/src/MoShiro.Agent/Chunks/ChunkAssembler.cs

@@ -18,6 +18,7 @@ public sealed class ChunkAssembler
     private readonly NodeId _agent;
     private readonly TimeProvider _time;
     private readonly INodeLog _log;
+    private readonly TaskTrafficReporter _traffic;
 
     public ChunkAssembler(StreamRegistry streams, MonoClock clock, AgentCounters counters, Outbox outbox, NodeId agent, TimeProvider time, INodeLog log)
     {
@@ -28,6 +29,7 @@ public sealed class ChunkAssembler
         _agent = agent;
         _time = time;
         _log = log;
+        _traffic = new TaskTrafficReporter(log, time);
     }
 
     /// <summary>
@@ -68,9 +70,20 @@ public sealed class ChunkAssembler
         return true;
     }
 
+    public Task RunTrafficReportingAsync(CancellationToken ct) => _traffic.RunAsync(ct);
+
+    public void FlushTraffic() => _traffic.Flush();
+
     /// <summary>Agent 自己采集的块（目标任务）直接发出。</summary>
     public void Emit(RawChunk chunk)
     {
-        _outbox.Enqueue(MoShiroChannels.RawChunk, chunk.Header.ChunkId.Value, RawChunkCodec.Encode(chunk));
+        byte[] payload = RawChunkCodec.Encode(chunk, out int uncompressedBytes);
+        _outbox.Enqueue(MoShiroChannels.RawChunk, chunk.Header.ChunkId.Value, payload);
+        _traffic.Record(
+            chunk.Header.TaskId?.Value ?? chunk.Header.TestId?.Value ?? "unknown",
+            chunk.Header.Kind,
+            chunk.Streams.Sum(stream => (long)stream.Count),
+            uncompressedBytes,
+            payload.Length);
     }
 }

backend/agent/src/MoShiro.Agent/Chunks/TaskTrafficReporter.cs

@@ -0,0 +1,107 @@
+using Fleet.Core;
+using Fleet.Rpc;
+
+namespace MoShiro.Agent;
+
+/// <summary>Aggregates per-task Agent → Backend chunk sizes into low-volume diagnostic logs.</summary>
+internal sealed class TaskTrafficReporter
+{
+    private static readonly TimeSpan s_window = TimeSpan.FromMinutes(1);
+
+    private readonly object _gate = new();
+    private readonly INodeLog _log;
+    private readonly TimeProvider _time;
+    private readonly Dictionary<(string Task, string Kind), Totals> _totals = [];
+    private long _windowStarted;
+
+    public TaskTrafficReporter(INodeLog log, TimeProvider time)
+    {
+        _log = log;
+        _time = time;
+        _windowStarted = time.GetTimestamp();
+    }
+
+    public void Record(string task, string kind, long records, int uncompressedBytes, int compressedBytes)
+    {
+        lock (_gate)
+        {
+            var key = (task, kind);
+            if (!_totals.TryGetValue(key, out Totals? totals))
+            {
+                totals = new Totals();
+                _totals.Add(key, totals);
+            }
+
+            totals.Chunks++;
+            totals.Records += records;
+            totals.UncompressedBytes += uncompressedBytes;
+            totals.CompressedBytes += compressedBytes;
+        }
+    }
+
+    public async Task RunAsync(CancellationToken ct)
+    {
+        using var timer = new PeriodicTimer(s_window, _time);
+        try
+        {
+            while (await timer.WaitForNextTickAsync(ct))
+            {
+                Flush();
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            // The final snapshot is flushed by the Agent shutdown path.
+        }
+    }
+
+    public void Flush()
+    {
+        List<Snapshot> completed;
+        long windowSeconds;
+        lock (_gate)
+        {
+            if (_totals.Count == 0)
+            {
+                _windowStarted = _time.GetTimestamp();
+                return;
+            }
+
+            TimeSpan elapsed = _time.GetElapsedTime(_windowStarted);
+            windowSeconds = Math.Max(1, (long)elapsed.TotalSeconds);
+            completed = [.. _totals.Select(pair => new Snapshot(pair.Key.Task, pair.Key.Kind, pair.Value.Chunks, pair.Value.Records, pair.Value.UncompressedBytes, pair.Value.CompressedBytes))];
+            _totals.Clear();
+            _windowStarted = _time.GetTimestamp();
+        }
+
+        foreach (Snapshot item in completed)
+        {
+            double savedPercent = item.UncompressedBytes == 0
+                ? 0
+                : 100d * (item.UncompressedBytes - item.CompressedBytes) / item.UncompressedBytes;
+            long queuedBytes = item.CompressedBytes + (long)item.Chunks * OutboxFrame.HeaderBytes;
+            _log.Info(
+                "task upload traffic",
+                ("link", "agent-backend"),
+                ("task", item.Task),
+                ("kind", item.Kind),
+                ("chunks", item.Chunks),
+                ("records", item.Records),
+                ("raw_bytes", item.UncompressedBytes),
+                ("compressed_bytes", item.CompressedBytes),
+                ("queued_bytes", queuedBytes),
+                ("saved_pct", Math.Round(savedPercent, 1)),
+                ("window_s", windowSeconds));
+        }
+    }
+
+    private sealed class Totals
+    {
+        public long Chunks;
+        public long Records;
+        public long UncompressedBytes;
+        public long CompressedBytes;
+    }
+
+    private sealed record Snapshot(string Task, string Kind, long Chunks, long Records, long UncompressedBytes, long CompressedBytes);
+}

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

@@ -3,7 +3,7 @@ using Fleet.Hosting;
 using MoShiro.Agent;
 
 TimeProvider time = TimeProvider.System;
-INodeLog log = new ConsoleNodeLog(time);
+INodeLog log = new BufferedNodeLog(new ConsoleNodeLog(time), time);
 
 if (PlatformCheck.VerifyCrypto() is { } problem)
 {

backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs

@@ -161,6 +161,7 @@ public sealed class TaskManager
             foreach ((ProbeSession probe, Dictionary<string, LineEntry> current) in _sessionLines)
             {
                 var next = new Dictionary<string, LineEntry>();
+                bool lineTasksChanged = current.Count != line.Count;
                 foreach ((string id, (TaskSpec task, TaskDecision decision)) in line)
                 {
                     LineEntry candidate = BuildLineEntry(task, decision, probe);
@@ -170,6 +171,7 @@ public sealed class TaskManager
                     }
                     else
                     {
+                        lineTasksChanged = true;
                         Start(candidate);
                         next[id] = candidate;
                     }
@@ -189,7 +191,13 @@ public sealed class TaskManager
                     current[id] = entry;
                 }
 
-                changes.Add((probe, SetFor(current)));
+                // The Agent receives a full task snapshot for peer/trust changes and target-only tasks too.
+                // Those changes do not affect a Probe's line blocks, so avoid retransmitting the same full
+                // line_task_set to every Probe. New sessions still get their initial snapshot in OpenSession.
+                if (lineTasksChanged)
+                {
+                    changes.Add((probe, SetFor(current)));
+                }
             }
         }