优化数据消耗

2026-09-09 00:11:03 +08:00
1 parent a7194dd287
commit c971049290
@@ -122,6 +122,10 @@ jobs:
        run: cargo fmt --manifest-path backend/Cargo.toml -- --check
      - name: Check Agent formatting
        run: cargo fmt --manifest-path agent/Cargo.toml -- --check
+      - name: Check telemetry formatting
+        run: cargo fmt --manifest-path shared/Cargo.toml -- --check
+      - name: Test telemetry codec
+        run: CARGO_TARGET_DIR=target/backend cargo test --locked --manifest-path shared/Cargo.toml
      - name: Lint backend
        run: CARGO_TARGET_DIR=target/backend cargo clippy --locked --manifest-path backend/Cargo.toml --all-targets -- -D warnings
      - name: Lint Agent
@@ -4,6 +4,7 @@
 /backend/target/
 /backend/build/
 /agent/target/
+/shared/target/
 /frontend/dist/
 /frontend/admin-dist/
 /release/
@@ -121,7 +121,9 @@ Linux systemd 下，请通过 SSH 或本机终端重装、手动更新或卸载

 远程执行是异步批量任务：只向当前连接的 Agent 下发命令，离线节点直接报告失败，重连后不会自动重发旧命令。多行脚本保留原始内容，执行结束后一次性返回输出；Agent 对单条命令设有 10 分钟执行上限。离开后台或退出登录不会停止已下发的命令。结果每 2 秒查询一次，持续 1 分钟后暂停自动刷新，可点击“刷新结果”继续查询；暂停查询不代表命令执行失败或已停止。

-实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，前台每秒合并展示最新样本，无新样本时保持原值。节点实时采样间隔默认为 1 秒；设置更长间隔时不会凭空生成每秒数据。历史保存间隔独立，默认 60 秒。
+实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，默认每 3 秒采集并上传一次，可设置为 3–60 秒。前台每秒合并展示已收到的最新样本，无新样本时保持原值。历史保存间隔独立，默认 60 秒。
+
+Agent 和前台的监控数据使用 gzip 压缩的二进制 WebSocket 消息，机器静态信息仅在连接建立或发生变化时上传。正常连接只发送新增样本，后端暂存后按历史保存间隔批量写入数据库并确认；Agent 保留未确认样本，断线后补传。页面访问不会改变节点上报频率。

 节点价格最低为 `0`，表示免费；不再接受负价格。是否隐藏节点仅由“隐藏节点”开关决定。

@@ -2,6 +2,12 @@
 # It is not intended for manual editing.
 version = 4

+[[package]]
+name = "adler2"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
+
 [[package]]
 name = "anstyle"
 version = "1.0.14"
@@ -145,6 +151,15 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "crc32fast"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8498c871161e1742aaa9d52551b2d6ebdd4c3d45a3be423e3728f33b955be550"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "crypto-common"
 version = "0.2.2"
@@ -209,6 +224,17 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890"

+[[package]]
+name = "flate2"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
+dependencies = [
+ "crc32fast",
+ "miniz_oxide",
+ "zlib-rs",
+]
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.2"
@@ -434,12 +460,23 @@ version = "2.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"

+[[package]]
+name = "miniz_oxide"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
+dependencies = [
+ "adler2",
+ "simd-adler32",
+]
+
 [[package]]
 name = "nodeflare-agent"
 version = "1.0.0"
 dependencies = [
 "clap",
 "httpdate",
+ "nodeflare-telemetry",
 "serde",
 "serde_json",
 "sha2",
@@ -449,6 +486,15 @@ dependencies = [
 "url",
 ]

+[[package]]
+name = "nodeflare-telemetry"
+version = "1.0.0"
+dependencies = [
+ "flate2",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "ntapi"
 version = "0.4.3"
@@ -671,6 +717,12 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"

+[[package]]
+name = "simd-adler32"
+version = "0.3.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea"
+
 [[package]]
 name = "smallvec"
 version = "1.16.0"
@@ -1218,6 +1270,12 @@ dependencies = [
 "syn 3.0.4",
 ]

+[[package]]
+name = "zlib-rs"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12"
+
 [[package]]
 name = "zmij"
 version = "1.0.23"
@@ -10,6 +10,7 @@ repository = "https://github.com/elysia62/NodeFlare"
 clap = { version = "4.6", default-features = false, features = ["derive", "std", "help", "usage", "error-context", "suggestions"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+nodeflare-telemetry = { path = "../shared" }
 sha2 = "0.11"
 httpdate = "1"
 ureq = { version = "3.4", default-features = false, features = ["json", "rustls"] }
@@ -2,7 +2,7 @@ use std::collections::VecDeque;

 use super::Report;

-pub(crate) const MAX_LIVE_BATCH_BYTES: usize = 768 * 1024;
+pub(crate) const MAX_LIVE_BATCH_BYTES: usize = nodeflare_telemetry::MAX_BATCH_BYTES;

 pub(crate) fn batch_len(queue: &VecDeque<Report>) -> usize {
    let mut encoded_bytes = 192_usize;
@@ -13,7 +13,7 @@ pub(crate) fn batch_len(queue: &VecDeque<Report>) -> usize {
        };
        let next_bytes = encoded_bytes
            .saturating_add(report_bytes.len())
-            .saturating_add(usize::from(count > 0));
+            .saturating_add(256);
        if next_bytes > MAX_LIVE_BATCH_BYTES {
            break;
        }
@@ -43,8 +43,9 @@ mod tests {
        let count = batch_len(&queue);
        assert!((1..10).contains(&count));
        let payload = crate::live_update_payload(
-            &queue.iter().take(count).cloned().collect::<Vec<_>>(),
+            queue.iter().take(count).cloned().collect::<Vec<_>>(),
            true,
+            &mut None,
        )
        .unwrap();
        assert!(payload.len() <= MAX_LIVE_BATCH_BYTES);
@@ -75,7 +76,7 @@ mod tests {
            if batch.is_empty() {
                break;
            }
-            let message = crate::live_update_payload(&batch, true).unwrap();
+            let message = crate::live_update_payload(batch.clone(), true, &mut None).unwrap();
            assert!(message.len() <= MAX_LIVE_BATCH_BYTES);
            persisted = batch.last().unwrap().timestamp;
            received.extend(batch.into_iter().map(|report| report.timestamp));
@@ -95,10 +96,10 @@ mod tests {
        let reports = crate::live_batch_after(&queue, 100);
        assert_eq!(reports.first().unwrap().timestamp, 101);
        for persist in [true, false] {
-            let message = crate::live_update_payload(&reports, persist).unwrap();
-            let value: serde_json::Value = serde_json::from_str(&message).unwrap();
+            let message = crate::live_update_payload(reports.clone(), persist, &mut None).unwrap();
+            let value: serde_json::Value = nodeflare_telemetry::decode(&message).unwrap();
            assert_eq!(value["persist"], persist);
-            assert_eq!(value.as_object().unwrap().len(), 4);
+            assert_eq!(value.as_object().unwrap().len(), 2);
        }
    }
 }
@@ -1,5 +1,7 @@
 mod live_batch;
 mod runtime_stats;
+use nodeflare_telemetry as telemetry;
+use telemetry::{DiskMetric, GpuMetric, LatencyResult, Report};

 use std::collections::{HashMap, HashSet, VecDeque};
 use std::env;
@@ -63,7 +65,6 @@ const MAX_PENDING_SPOOL_BYTES: u64 = 64 * 1024 * 1024;
 const MAX_PENDING_SPOOL_LINE_BYTES: usize = 1024 * 1024;
 const LIVE_ACK_READ_TIMEOUT: Duration = Duration::from_secs(5);
 const LIVE_HINT_READ_TIMEOUT: Duration = Duration::from_millis(10);
-const LIVE_REPORT_DIVISOR: u64 = 15;
 const BASIC_INFO_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);
 const SLOW_METRICS_REFRESH_INTERVAL: Duration = Duration::from_secs(5);
 const PUBLIC_IP_REFRESH_INTERVAL: Duration = Duration::from_secs(10 * 60);
@@ -122,14 +123,6 @@ struct LatencyTask {
    interval_seconds: u64,
 }

-#[derive(Debug, Clone, Deserialize, Serialize)]
-struct LatencyResult {
-    task_id: String,
-    timestamp: i64,
-    latency_ms: f64,
-    packet_loss: f64,
-}
-
 struct LatencyExecutor {
    task_tx: mpsc::Sender<LatencyTask>,
    result_rx: mpsc::Receiver<LatencyResult>,
@@ -383,10 +376,7 @@ impl LiveSender {
        stats: Arc<runtime_stats::RuntimeStats>,
    ) -> Result<Self> {
        let pending = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));
-        let send_interval = Arc::new(Mutex::new(live_batch_interval(
-            config.report_interval,
-            config.collect_interval,
-        )));
+        let send_interval = Arc::new(Mutex::new(live_batch_interval(config.collect_interval)));
        let persisted_through = Arc::new(AtomicI64::new(0));
        let remote_config = Arc::new(Mutex::new(None));
        let endpoint = live_endpoint(&config.endpoint)?;
@@ -433,9 +423,9 @@ impl LiveSender {
        }
    }

-    fn set_send_interval(&self, report_interval: u64, collect_interval: u64) {
+    fn set_send_interval(&self, collect_interval: u64) {
        if let Ok(mut interval) = self.send_interval.lock() {
-            *interval = live_batch_interval(report_interval, collect_interval);
+            *interval = live_batch_interval(collect_interval);
        }
        self.pending.1.notify_one();
    }
@@ -452,78 +442,8 @@ impl LiveSender {
    }
 }

-fn live_batch_interval(report_interval: u64, collect_interval: u64) -> Duration {
-    let seconds = report_interval
-        .clamp(15, 3600)
-        .div_ceil(LIVE_REPORT_DIVISOR)
-        .clamp(1, 60)
-        .max(collect_interval.clamp(1, 60));
-    Duration::from_secs(seconds)
-}
-
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
-struct DiskMetric {
-    name: String,
-    mount_point: String,
-    used: i64,
-    total: i64,
-    read_bps: f64,
-    write_bps: f64,
-    read_iops: f64,
-    write_iops: f64,
-    await_ms: f64,
-    utilization: f64,
-}
-
-#[derive(Debug, Clone, Deserialize, Serialize)]
-struct GpuMetric {
-    model: String,
-    usage: Option<f64>,
-    memory_used: i64,
-    memory_total: i64,
-}
-
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
-struct Report {
-    timestamp: i64,
-    cpu: f64,
-    load1: f64,
-    load5: f64,
-    load15: f64,
-    mem_used: i64,
-    mem_total: i64,
-    swap_used: i64,
-    swap_total: i64,
-    disk_used: i64,
-    disk_total: i64,
-    net_in: f64,
-    net_out: f64,
-    net_rx_total: i64,
-    net_tx_total: i64,
-    uptime: i64,
-    processes: i64,
-    tcp_connections: i64,
-    udp_connections: i64,
-    cpu_cores: i64,
-    cpu_model: String,
-    os: String,
-    kernel: String,
-    arch: String,
-    virtualization: String,
-    gpu_usage: f64,
-    gpu_model: String,
-    agent_version: String,
-    ip_v4: String,
-    ip_v6: String,
-    disk_read_bps: f64,
-    disk_write_bps: f64,
-    disk_read_iops: f64,
-    disk_write_iops: f64,
-    disk_await_ms: f64,
-    disk_utilization: f64,
-    disks: Vec<DiskMetric>,
-    gpus: Vec<GpuMetric>,
-    latency_results: Vec<LatencyResult>,
+fn live_batch_interval(collect_interval: u64) -> Duration {
+    Duration::from_secs(collect_interval.clamp(telemetry::MIN_COLLECT_INTERVAL, 60))
 }

 #[derive(Debug, Default, Clone)]
@@ -1720,7 +1640,7 @@ fn runtime_config(options: &CliOptions) -> Result<RuntimeConfig> {
        token,
        endpoint: endpoint.trim_end_matches('/').to_string(),
        report_interval: interval,
-        collect_interval: 1,
+        collect_interval: telemetry::MIN_COLLECT_INTERVAL,
        network_interface: String::new(),
        agent_mirror: String::new(),
        auto_update: true,
@@ -1737,12 +1657,6 @@ fn live_endpoint(endpoint: &str) -> Result<String> {
    Ok(url.to_string())
 }

-fn report_batch_id(reports: &[Report]) -> String {
-    let encoded = serde_json::to_vec(reports).unwrap_or_default();
-    let digest = Sha256::digest(encoded);
-    digest.iter().map(|byte| format!("{byte:02x}")).collect()
-}
-
 type LiveSocket = tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<TcpStream>>;

 fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option<i64>)> {
@@ -1853,10 +1767,6 @@ struct LiveAck {
    persisted_through_ts: i64,
    #[serde(rename = "nextPersistAfterMs")]
    next_persist_after_ms: u64,
-    #[serde(rename = "nextWssReportAfterMs")]
-    next_wss_report_after_ms: u64,
-    #[serde(rename = "realtimeHint")]
-    realtime_hint: bool,
 }

 #[derive(Debug, Deserialize)]
@@ -2463,11 +2373,6 @@ fn flush_remote_results(socket: &mut LiveSocket, executor: &mut RemoteExecutor)
    Ok(())
 }

-fn ack_wss_interval(ack: &LiveAck) -> Duration {
-    Duration::from_millis(ack.next_wss_report_after_ms)
-        .clamp(Duration::from_secs(1), Duration::from_secs(3600))
-}
-
 fn ack_persist_interval(ack: &LiveAck) -> Duration {
    Duration::from_millis(ack.next_persist_after_ms)
        .clamp(Duration::from_secs(1), Duration::from_secs(3600))
@@ -2583,7 +2488,7 @@ fn wait_for_live_ack(
        }
        set_live_read_timeout(socket, Some(remaining))?;
        match read_live_ack(socket)? {
-            LiveRead::Ack(ack) if !ack.realtime_hint => {
+            LiveRead::Ack(ack) => {
                return Ok(LiveRead::Ack(ack));
            }
            LiveRead::Config(config) => {
@@ -2598,24 +2503,30 @@ fn wait_for_live_ack(
                remote_executor.acknowledge(&task_id);
            }
            LiveRead::Closed => return Ok(LiveRead::Closed),
-            LiveRead::Ack(_) | LiveRead::Pending => {}
+            LiveRead::Pending => {}
        }
    }
 }

-fn live_update_payload(reports: &[Report], persist: bool) -> Result<String> {
-    let message = serde_json::json!({
-        "type": "update",
-        "batchId": report_batch_id(reports),
-        "samples": reports,
-        "persist": persist,
-    });
-    Ok(serde_json::to_string(&message)?)
+fn live_update_payload(
+    reports: Vec<Report>,
+    persist: bool,
+    info: &mut Option<telemetry::Info>,
+) -> Result<Vec<u8>> {
+    let mut next_info = info.clone();
+    let payload = telemetry::encode(&telemetry::Update::from_reports(
+        reports,
+        persist,
+        &mut next_info,
+    ))?;
+    *info = next_info;
+    Ok(payload)
 }

 fn observe_persisted_through(target: &AtomicI64, ack: &LiveAck) {
-    let _ = ack.persisted;
-    target.fetch_max(ack.persisted_through_ts.max(0), Ordering::AcqRel);
+    if ack.persisted && !ack.persistence_error {
+        target.fetch_max(ack.persisted_through_ts.max(0), Ordering::AcqRel);
+    }
 }

 fn prune_live_queue(pending: &Arc<(Mutex<VecDeque<Report>>, Condvar)>, persisted_through: i64) {
@@ -2647,9 +2558,11 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
        stats,
    } = worker;
    let mut socket: Option<LiveSocket> = None;
-    let mut wss_interval = configured_interval
-        .lock()
-        .map_or(Duration::from_secs(1), |interval| *interval);
+    let mut wss_interval = configured_interval.lock().map_or(
+        Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL),
+        |interval| *interval,
+    );
+    let mut info = None;
    let mut accepted_through = persisted_through.load(Ordering::Acquire);
    let mut next_send_at = Instant::now();
    let mut next_probe_at = Instant::now();
@@ -2671,6 +2584,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                        continue;
                    }
                    socket = Some(connected);
+                    info = None;
                    remote_executor.reset_delivery();
                    accepted_through = persisted_through.load(Ordering::Acquire);
                    next_send_at = Instant::now();
@@ -2700,17 +2614,16 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
        let (batch, persist, more_pending) = loop {
            let now = Instant::now();
            let persisted = persisted_through.load(Ordering::Acquire);
-            if now >= next_probe_at {
-                let persistence_batch = live_batch_after(&queue, persisted);
-                if !persistence_batch.is_empty() {
-                    let last = persistence_batch.last().unwrap().timestamp;
-                    let more_pending = queue.iter().any(|report| report.timestamp > last);
-                    break (persistence_batch, true, more_pending);
-                }
-            }
            let unsent = live_batch_after(&queue, accepted_through);
-            if !unsent.is_empty() && now >= next_send_at {
-                break (unsent, false, false);
+            let persist =
+                now >= next_probe_at && queue.iter().any(|report| report.timestamp > persisted);
+            if persist || (!unsent.is_empty() && now >= next_send_at) {
+                let last = unsent
+                    .last()
+                    .map_or(accepted_through, |report| report.timestamp);
+                let more_pending = queue.iter().any(|report| report.timestamp > last);
+                // A commit can be empty: samples already sent remain buffered by the server.
+                break (unsent, persist || more_pending, more_pending);
            }

            let wake_at = if !unsent.is_empty() {
@@ -2738,10 +2651,6 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                } else {
                    match read_live_ack(connected) {
                        Ok(LiveRead::Closed) => drop_socket = true,
-                        Ok(LiveRead::Ack(ack)) if ack.realtime_hint => {
-                            wss_interval = ack_wss_interval(&ack);
-                            next_send_at = Instant::now() + wss_interval;
-                        }
                        Ok(LiveRead::Ack(ack)) => {
                            observe_persisted_through(&persisted_through, &ack);
                            if ack.persistence_error {
@@ -2750,8 +2659,6 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                            } else {
                                prune_live_queue(&pending, ack.persisted_through_ts);
                                accepted_through = accepted_through.max(ack.persisted_through_ts);
-                                wss_interval = ack_wss_interval(&ack);
-                                next_send_at = Instant::now() + wss_interval;
                                next_probe_at = Instant::now() + ack_persist_interval(&ack);
                            }
                        }
@@ -2792,10 +2699,15 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
        drop(queue);

        if let Ok(interval) = configured_interval.lock() {
-            wss_interval = (*interval).clamp(Duration::from_secs(1), Duration::from_secs(60));
+            wss_interval = (*interval).clamp(
+                Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL),
+                Duration::from_secs(60),
+            );
        }

-        let payload = match live_update_payload(&batch, persist) {
+        let batch_last = batch.last().map(|report| report.timestamp);
+        let batch_len = batch.len();
+        let payload = match live_update_payload(batch, persist, &mut info) {
            Ok(payload) => payload,
            Err(error) => {
                eprintln!("live payload encode failed: {error}");
@@ -2805,7 +2717,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
        };
        let sent = socket
            .as_mut()
-            .is_some_and(|socket| socket.send(Message::Text(payload.into())).is_ok());
+            .is_some_and(|socket| socket.send(Message::Binary(payload.into())).is_ok());
        if !sent {
            socket = None;
            accepted_through = persisted_through.load(Ordering::Acquire);
@@ -2814,7 +2726,14 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
            thread::sleep(LIVE_RECONNECT_DELAY);
            continue;
        }
-        stats.live_batch_sent(batch.len());
+        stats.live_batch_sent(batch_len);
+        if let Some(timestamp) = batch_last {
+            accepted_through = accepted_through.max(timestamp);
+        }
+        next_send_at = Instant::now() + wss_interval;
+        if !persist {
+            continue;
+        }

        let mut drop_socket = false;
        if let Some(socket) = socket.as_mut() {
@@ -2827,18 +2746,11 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                        drop_socket = true;
                    } else {
                        prune_live_queue(&pending, ack.persisted_through_ts);
-                        if let Some(timestamp) = batch.iter().map(|report| report.timestamp).max() {
-                            accepted_through = accepted_through.max(timestamp);
-                        }
                        accepted_through = accepted_through.max(ack.persisted_through_ts);
-                        wss_interval = ack_wss_interval(&ack);
-                        next_send_at = Instant::now() + wss_interval;
                        next_probe_at = Instant::now() + ack_persist_interval(&ack);
-                        if persist
-                            && more_pending
-                            && batch
-                                .last()
-                                .is_some_and(|report| ack.persisted_through_ts >= report.timestamp)
+                        if more_pending
+                            && batch_last
+                                .is_some_and(|timestamp| ack.persisted_through_ts >= timestamp)
                        {
                            next_probe_at = Instant::now();
                        }
@@ -2874,7 +2786,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
 fn apply_remote(config: &mut RuntimeConfig, remote: &RemoteConfig) -> bool {
    if valid_sample_schedule(remote.report_interval, remote.collect_interval) {
        config.report_interval = remote.report_interval;
-        config.collect_interval = remote.collect_interval;
+        config.collect_interval = remote.collect_interval.max(telemetry::MIN_COLLECT_INTERVAL);
    }
    config
        .network_interface
@@ -3380,7 +3292,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> {
                if config.collect_interval != previous_collect_interval
                    || config.report_interval != previous_report_interval
                {
-                    live.set_send_interval(config.report_interval, config.collect_interval);
+                    live.set_send_interval(config.collect_interval);
                }
            }
        }
@@ -3542,7 +3454,7 @@ mod tests {
        PUBLIC_IP_STALE_AFTER, PublicIpValue, REMOTE_RESULT_OUTPUT_BYTES,
        REMOTE_STREAM_OUTPUT_BYTES, RemoteExecutor, RemoteTaskJournalEntry, RemoteTaskMessage,
        Report, TaskResultMessage, UPDATE_CHECK_JITTER_MAX_SECONDS, ack_persist_interval,
-        ack_wss_interval, advance_deadline, clock_offset_from_http_date, corrected_timestamp,
+        advance_deadline, clock_offset_from_http_date, corrected_timestamp,
        execute_remote_task_with_timeout, gpu_name_from_uevent, is_public_probe_ip, live_endpoint,
        live_update_payload, monotonic_report_timestamp, normalized_version, parse_lspci_gpu_names,
        parse_pciconf_gpu_names, parse_probe_target, parse_public_ip,
@@ -3693,24 +3605,20 @@ mod tests {
    }

    #[test]
-    fn accepts_server_realtime_ack_interval() {
+    fn accepts_durable_ack_interval_and_limits_upload_frequency() {
        let ack: LiveAck = serde_json::from_str(
-            r#"{"type":"ack","ts":100,"persisted":false,"persistenceError":false,"persistedThroughTs":90,"nextPersistAfterMs":60000,"nextWssReportAfterMs":5000,"realtimeHint":false}"#,
+            r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":60000}"#,
        )
        .unwrap();
-        assert_eq!(ack_wss_interval(&ack), Duration::from_secs(5));
        assert_eq!(ack_persist_interval(&ack), Duration::from_secs(60));
-        let slow: LiveAck = serde_json::from_str(
-            r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":60000,"nextWssReportAfterMs":1,"realtimeHint":true}"#,
-        )
-        .unwrap();
-        assert_eq!(ack_wss_interval(&slow), Duration::from_secs(1));
-        assert!(slow.realtime_hint);
        let idle: LiveAck = serde_json::from_str(
-            r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":120000,"nextWssReportAfterMs":3600000,"realtimeHint":false}"#,
+            r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":120000}"#,
        )
        .unwrap();
        assert_eq!(ack_persist_interval(&idle), Duration::from_secs(120));
+        assert_eq!(super::live_batch_interval(1), Duration::from_secs(3));
+        assert_eq!(super::live_batch_interval(3), Duration::from_secs(3));
+        assert_eq!(super::live_batch_interval(15), Duration::from_secs(15));
    }

    #[cfg(unix)]
@@ -3873,7 +3781,7 @@ mod tests {
    #[test]
    fn encodes_realtime_samples_as_a_batch() {
        let payload = live_update_payload(
-            &[
+            vec![
                Report {
                    timestamp: 10,
                    cpu: 20.0,
@@ -3886,13 +3794,13 @@ mod tests {
                },
            ],
            false,
+            &mut None,
        )
        .unwrap();
-        let value: serde_json::Value = serde_json::from_str(&payload).unwrap();
-        assert_eq!(value["type"], "update");
+        let value: serde_json::Value = nodeflare_telemetry::decode(&payload).unwrap();
        assert_eq!(value["persist"], false);
        assert_eq!(value["samples"].as_array().unwrap().len(), 2);
-        assert_eq!(value["samples"][1]["cpu"], 30.0);
+        assert_eq!(value["samples"][1]["metrics"]["cpu"], 30.0);
    }

    #[test]
@@ -2,6 +2,12 @@
 # It is not intended for manual editing.
 version = 4

+[[package]]
+name = "adler2"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
+
 [[package]]
 name = "aho-corasick"
 version = "1.1.5"
@@ -538,6 +544,8 @@ version = "1.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
 dependencies = [
+ "crc32fast",
+ "miniz_oxide",
 "zlib-rs",
 ]

@@ -1155,6 +1163,16 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"

+[[package]]
+name = "miniz_oxide"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
+dependencies = [
+ "adler2",
+ "simd-adler32",
+]
+
 [[package]]
 name = "mio"
 version = "1.2.3"
@@ -1178,6 +1196,7 @@ dependencies = [
 "futures-util",
 "hex",
 "ipnet",
+ "nodeflare-telemetry",
 "pbkdf2",
 "rand 0.10.2",
 "reqwest",
@@ -1200,6 +1219,15 @@ dependencies = [
 "zip",
 ]

+[[package]]
+name = "nodeflare-telemetry"
+version = "1.0.0"
+dependencies = [
+ "flate2",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "nu-ansi-term"
 version = "0.50.3"
@@ -17,6 +17,7 @@ tokio-util = { version = "0.7", features = ["io"] }
 tower-http = { version = "0.7.1", features = ["trace"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+nodeflare-telemetry = { path = "../shared" }
 sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "tls-rustls-ring", "any", "sqlite-bundled", "postgres", "migrate", "macros"] }
 toml = "1.1.5"
 tracing = "0.1"
@@ -43,7 +43,6 @@ pub struct AgentIdentity {
    pub server_id: String,
    pub hidden: bool,
    pub report_interval: i64,
-    pub collect_interval: i64,
    pub reset_day: i64,
    pub rx_correction: i64,
    pub tx_correction: i64,
@@ -58,10 +57,10 @@ pub struct PersistResult {
 }

 #[derive(Debug, Clone, Default)]
-struct TrafficState {
+pub(crate) struct TrafficState {
    cycle_key: i64,
    reset_day: i64,
-    timestamp: i64,
+    pub timestamp: i64,
    raw_rx: i64,
    raw_tx: i64,
    used_rx: i64,
@@ -276,6 +275,14 @@ fn aggregate_history(reports: &[AgentReport], interval: i64) -> Vec<HistoryRow>
 }

 pub async fn list_servers(db: &Database, include_hidden: bool) -> Result<Vec<ServerView>> {
+    list_servers_with_live(db, include_hidden, &HashMap::new()).await
+}
+
+pub async fn list_servers_with_live(
+    db: &Database,
+    include_hidden: bool,
+    live: &HashMap<String, AgentReport>,
+) -> Result<Vec<ServerView>> {
    let statement = if include_hidden {
        "SELECT s.id, s.name, s.region, s.group_name, s.tags, s.hidden, s.expires_at, \
         s.traffic_limit, s.traffic_limit_type, s.price, s.billing_cycle, s.currency, \
@@ -296,17 +303,26 @@ pub async fn list_servers(db: &Database, include_hidden: bool) -> Result<Vec<Ser
         WHERE s.hidden=0 ORDER BY s.sort_order, s.created_at"
    };
    let rows = sqlx::query(statement).fetch_all(db.pool()).await?;
-    let latency = latest_latency_map(db).await?;
+    let latency = latest_latency_map(db, live).await?;
    rows.into_iter()
        .map(|row| {
            let id: String = row.try_get("id")?;
            let latest_json: Option<String> = row.try_get("latest_json")?;
-            let report = latest_json
+            let persisted_report = latest_json
                .as_deref()
                .filter(|value| !value.is_empty() && *value != "{}")
                .and_then(|value| serde_json::from_str::<AgentReport>(value).ok());
-            let value = |read: fn(&AgentReport) -> f64| report.as_ref().map(read);
-            let integer = |read: fn(&AgentReport) -> i64| report.as_ref().map(read);
+            let report = live
+                .get(&id)
+                .filter(|current| {
+                    current.timestamp
+                        >= persisted_report
+                            .as_ref()
+                            .map_or(0, |report| report.timestamp)
+                })
+                .or(persisted_report.as_ref());
+            let value = |read: fn(&AgentReport) -> f64| report.map(read);
+            let integer = |read: fn(&AgentReport) -> i64| report.map(read);
            let text = |read: fn(&AgentReport) -> &String| report.as_ref().map(|r| read(r).clone());
            Ok(ServerView {
                id: id.clone(),
@@ -328,7 +344,9 @@ pub async fn list_servers(db: &Database, include_hidden: bool) -> Result<Vec<Ser
                network_interface: row.try_get("network_interface")?,
                reset_day: row.try_get("reset_day")?,
                report_interval: row.try_get("report_interval")?,
-                collect_interval: row.try_get("collect_interval")?,
+                collect_interval: row
+                    .try_get::<i64, _>("collect_interval")?
+                    .max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64),
                rx_correction: row.try_get("rx_correction")?,
                tx_correction: row.try_get("tx_correction")?,
                agent_mirror: row.try_get("agent_mirror")?,
@@ -544,7 +562,7 @@ pub async fn server_name(db: &Database, id: &str) -> Result<Option<String>> {
 pub async fn agent_identity(db: &Database, token: &str) -> Result<Option<AgentIdentity>> {
    let token_hash = auth::token_hash(token);
    let row = sqlx::query(db.sql(
-        "SELECT id, hidden, report_interval, collect_interval, reset_day, rx_correction, \
+        "SELECT id, hidden, report_interval, reset_day, rx_correction, \
         tx_correction FROM servers WHERE token_hash=? OR EXISTS (\
         SELECT 1 FROM server_install_tokens WHERE server_id=servers.id AND token_hash=?\
         )",
@@ -558,7 +576,6 @@ pub async fn agent_identity(db: &Database, token: &str) -> Result<Option<AgentId
            server_id: row.try_get("id")?,
            hidden: row.try_get::<i64, _>("hidden")? != 0,
            report_interval: row.try_get("report_interval")?,
-            collect_interval: row.try_get("collect_interval")?,
            reset_day: row.try_get("reset_day")?,
            rx_correction: row.try_get("rx_correction")?,
            tx_correction: row.try_get("tx_correction")?,
@@ -582,7 +599,7 @@ pub async fn agent_config(db: &Database, id: &str) -> Result<Option<serde_json::
    let tasks = tasks_for_server(db, id).await?;
    Ok(Some(serde_json::json!({
        "report_interval": row.try_get::<i64, _>("report_interval")?,
-        "collect_interval": row.try_get::<i64, _>("collect_interval")?,
+        "collect_interval": row.try_get::<i64, _>("collect_interval")?.max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64),
        "network_interface": row.try_get::<String, _>("network_interface")?,
        "agent_mirror": row.try_get::<String, _>("agent_mirror")?,
        "auto_update": row.try_get::<i64, _>("auto_update")? != 0,
@@ -596,7 +613,6 @@ pub async fn save_agent_batch(
    batch_id: &str,
    reports: &[AgentReport],
    remote_ip: &str,
-    persist: bool,
 ) -> Result<PersistResult> {
    let mut reports = reports.to_vec();
    reports.sort_by_key(|report| report.timestamp);
@@ -615,16 +631,12 @@ pub async fn save_agent_batch(
        anyhow::bail!("report batch exceeds 720 samples");
    }

-    let needs_write = persist
-        || reports
-            .iter()
-            .any(|report| !report.latency_results.is_empty());
-    let mut transaction = if db.is_postgres() || !needs_write {
+    let mut transaction = if db.is_postgres() {
        db.pool().begin().await?
    } else {
        db.pool().begin_with("BEGIN IMMEDIATE").await?
    };
-    if db.is_postgres() && needs_write {
+    if db.is_postgres() {
        sqlx::query("SELECT id FROM servers WHERE id=$1 FOR UPDATE")
            .bind(&identity.server_id)
            .fetch_one(&mut *transaction)
@@ -666,29 +678,7 @@ pub async fn save_agent_batch(
    for report in &mut reports {
        apply_traffic(report, &mut traffic, identity);
    }
-    let latest = reports.last().context("report batch became empty")?;
    let report_interval = identity.report_interval.clamp(15, 3600);
-    let next_persist_at = persisted_before.saturating_add(report_interval);
-    // Only a complete persistence batch may advance the durable watermark.
-    if !persist {
-        let next_persist_after_ms =
-            next_persist_at.saturating_sub(latest.timestamp).max(1) as u64 * 1000;
-        if reports
-            .iter()
-            .any(|report| !report.latency_results.is_empty())
-        {
-            save_latency_rows(db, &mut transaction, &identity.server_id, &reports).await?;
-            transaction.commit().await?;
-        } else {
-            transaction.rollback().await?;
-        }
-        return Ok(PersistResult {
-            reports,
-            persisted: false,
-            persisted_through: persisted_before,
-            next_persist_after_ms,
-        });
-    }

    let history_rows = aggregate_history(&reports, report_interval);
    save_history_rows(db, &mut transaction, &identity.server_id, &history_rows).await?;
@@ -731,7 +721,7 @@ pub async fn save_agent_batch(
    })
 }

-fn valid_agent_report(report: &AgentReport, current: i64) -> bool {
+pub(crate) fn valid_agent_report(report: &AgentReport, current: i64) -> bool {
    fn finite_between(value: f64, minimum: f64, maximum: f64) -> bool {
        value.is_finite() && (minimum..=maximum).contains(&value)
    }
@@ -849,7 +839,18 @@ async fn load_traffic_state(
    .map_err(Into::into)
 }

-fn apply_traffic(report: &mut AgentReport, state: &mut TrafficState, identity: &AgentIdentity) {
+pub(crate) async fn agent_traffic_state(db: &Database, server_id: &str) -> Result<TrafficState> {
+    let mut transaction = db.pool().begin().await?;
+    let state = load_traffic_state(db, &mut transaction, server_id).await?;
+    transaction.rollback().await?;
+    Ok(state)
+}
+
+pub(crate) fn apply_traffic(
+    report: &mut AgentReport,
+    state: &mut TrafficState,
+    identity: &AgentIdentity,
+) {
    if report.timestamp <= state.timestamp {
        report.net_rx_total = state.used_rx.saturating_add(identity.rx_correction).max(0);
        report.net_tx_total = state.used_tx.saturating_add(identity.tx_correction).max(0);
@@ -1171,12 +1172,13 @@ pub async fn history(db: &Database, server_id: &str, hours: i64) -> Result<Vec<H
         MAX(disk_utilization) AS disk_utilization FROM ( \
           SELECT (timestamp / ?) * ? AS bucket_timestamp, metric_history.* \
           FROM metric_history WHERE server_id=? \
-             AND last_timestamp>=? \
+             AND timestamp>=? AND last_timestamp>=? \
         ) samples GROUP BY bucket_timestamp ORDER BY bucket_timestamp",
    ))
    .bind(bucket)
    .bind(bucket)
    .bind(server_id)
+    .bind(since.saturating_sub(3600))
    .bind(since)
    .fetch_all(db.pool())
    .await?;
@@ -1327,9 +1329,23 @@ fn days_in_month(year: i64, month: i64) -> i64 {
    }
 }

-async fn latest_latency_map(db: &Database) -> Result<HashMap<String, Vec<LatencySample>>> {
+async fn latest_latency_map(
+    db: &Database,
+    live: &HashMap<String, AgentReport>,
+) -> Result<HashMap<String, Vec<LatencySample>>> {
+    let mut live_latency = HashMap::<(&str, &str), &AgentLatencyResult>::new();
+    for (server_id, report) in live {
+        for value in &report.latency_results {
+            let current = live_latency
+                .entry((server_id, &value.task_id))
+                .or_insert(value);
+            if value.timestamp > current.timestamp {
+                *current = value;
+            }
+        }
+    }
    let rows = sqlx::query(
-        "SELECT a.server_id, t.id AS task_id, t.name, t.task_type, t.target, t.port, \
+        "SELECT a.server_id, a.assigned_at, t.id AS task_id, t.name, t.task_type, t.target, t.port, \
         lr.timestamp, lr.latency_ms, lr.packet_loss FROM latency_task_servers a \
         JOIN latency_tasks t ON t.id=a.task_id LEFT JOIN latency_results lr \
         ON lr.task_id=a.task_id AND lr.server_id=a.server_id AND lr.timestamp=( \
@@ -1343,24 +1359,31 @@ async fn latest_latency_map(db: &Database) -> Result<HashMap<String, Vec<Latency
    let mut result = HashMap::<String, Vec<LatencySample>>::new();
    for row in rows {
        let server_id: String = row.try_get("server_id")?;
-        result
-            .entry(server_id.clone())
-            .or_default()
-            .push(LatencySample {
-                task_id: row.try_get("task_id")?,
-                server_id,
-                name: row.try_get("name")?,
-                task_type: row.try_get("task_type")?,
-                target: row.try_get("target")?,
-                port: row.try_get("port")?,
-                timestamp: row
-                    .try_get::<Option<i64>, _>("timestamp")?
-                    .unwrap_or_default(),
-                latency_ms: row.try_get::<Option<f64>, _>("latency_ms")?.unwrap_or(-1.0),
-                packet_loss: row
-                    .try_get::<Option<f64>, _>("packet_loss")?
-                    .unwrap_or(-1.0),
-            });
+        let mut point = LatencySample {
+            task_id: row.try_get("task_id")?,
+            server_id: server_id.clone(),
+            name: row.try_get("name")?,
+            task_type: row.try_get("task_type")?,
+            target: row.try_get("target")?,
+            port: row.try_get("port")?,
+            timestamp: row
+                .try_get::<Option<i64>, _>("timestamp")?
+                .unwrap_or_default(),
+            latency_ms: row.try_get::<Option<f64>, _>("latency_ms")?.unwrap_or(-1.0),
+            packet_loss: row
+                .try_get::<Option<f64>, _>("packet_loss")?
+                .unwrap_or(-1.0),
+        };
+        let assigned_at = row.try_get::<i64, _>("assigned_at")?;
+        if let Some(current) = live_latency
+            .get(&(server_id.as_str(), point.task_id.as_str()))
+            .filter(|value| value.timestamp >= point.timestamp && value.timestamp >= assigned_at)
+        {
+            point.timestamp = current.timestamp;
+            point.latency_ms = current.latency_ms;
+            point.packet_loss = current.packet_loss;
+        }
+        result.entry(server_id).or_default().push(point);
    }
    Ok(result)
 }
@@ -2672,17 +2695,17 @@ mod tests {
            sample(start + 1, 90.0),
            sample(start + 2, 20.0),
        ];
-        let first = save_agent_batch(&db, &identity, "first", &reports[..2], "", true)
+        let first = save_agent_batch(&db, &identity, "first", &reports[..2], "")
            .await
            .unwrap();
        assert_eq!(first.persisted_through, start + 1);
        // Reconnect with an overlapping batch: only the unacknowledged sample is added.
-        let second = save_agent_batch(&db, &identity, "overlap", &reports[1..], "", true)
+        let second = save_agent_batch(&db, &identity, "overlap", &reports[1..], "")
            .await
            .unwrap();
        assert_eq!(second.persisted_through, start + 2);
        for id in ["overlap", "another-retry"] {
-            let retry = save_agent_batch(&db, &identity, id, &reports, "", true)
+            let retry = save_agent_batch(&db, &identity, id, &reports, "")
                .await
                .unwrap();
            assert!(retry.reports.is_empty());
@@ -2712,20 +2735,27 @@ mod tests {
        let reports = (0..121)
            .map(|offset| sample(start + offset, 25.0))
            .collect::<Vec<_>>();
-        save_agent_batch(&db, &identity, "initial", &reports[..1], "", true)
+        save_agent_batch(&db, &identity, "initial", &reports[..1], "")
            .await
            .unwrap();
-        let live = save_agent_batch(&db, &identity, "live", &reports[120..], "", false)
+        let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id)
            .await
            .unwrap();
-        assert!(!live.persisted);
-        assert_eq!(live.persisted_through, start);
+        let live = buffer
+            .receive(&db, &identity, "", reports[120..].to_vec(), false)
+            .await
+            .unwrap();
+        assert!(live.acknowledgement.is_none());
+        assert_eq!(live.latest.unwrap().timestamp, start + 120);
+        assert_eq!(
+            list_servers(&db, true).await.unwrap()[0].timestamp,
+            Some(start)
+        );
        // Even a payload-size-limited batch shorter than report_interval must commit.
        for (index, batch) in reports[1..].chunks(3).enumerate() {
-            let result =
-                save_agent_batch(&db, &identity, &format!("batch-{index}"), batch, "", true)
-                    .await
-                    .unwrap();
+            let result = save_agent_batch(&db, &identity, &format!("batch-{index}"), batch, "")
+                .await
+                .unwrap();
            assert!(result.persisted);
            assert_eq!(result.persisted_through, batch.last().unwrap().timestamp);
        }
@@ -2738,11 +2768,182 @@ mod tests {
        assert_eq!(count, 121);
    }

+    #[tokio::test]
+    async fn latest_latency_cache_respects_probe_assignment_boundaries() {
+        let (db, identity) = test_server().await;
+        let current = now();
+        let task_id = create_latency_task(
+            &db,
+            &LatencyTaskInput {
+                name: "Cache test".into(),
+                task_type: "tcp".into(),
+                target: "example.com".into(),
+                port: Some(443),
+                interval_seconds: 60,
+                default_enabled: false,
+                server_ids: vec![identity.server_id.clone()],
+            },
+        )
+        .await
+        .unwrap();
+        let live = HashMap::from([(
+            identity.server_id.clone(),
+            AgentReport {
+                timestamp: current,
+                latency_results: vec![AgentLatencyResult {
+                    task_id: task_id.clone(),
+                    timestamp: current,
+                    latency_ms: 28.4,
+                    packet_loss: 0.0,
+                }],
+                ..AgentReport::default()
+            },
+        )]);
+        assert_eq!(
+            list_servers_with_live(&db, true, &live).await.unwrap()[0].latency[0].latency_ms,
+            28.4
+        );
+        sqlx::query("UPDATE latency_task_servers SET assigned_at=? WHERE task_id=?")
+            .bind(current + 1)
+            .bind(&task_id)
+            .execute(db.pool())
+            .await
+            .unwrap();
+        let latest = list_servers_with_live(&db, true, &live)
+            .await
+            .unwrap()
+            .remove(0);
+        assert_eq!(latest.latency[0].timestamp, 0);
+        assert_eq!(latest.latency[0].latency_ms, -1.0);
+    }
+
+    #[tokio::test]
+    async fn live_buffer_commits_without_reupload_and_replays_only_unconfirmed_samples() {
+        use crate::websocket::ingest::AgentBuffer;
+        let (db, identity) = test_server().await;
+        let start = now() - 120;
+        let first = sample(start, 10.0);
+        let second = sample(start + 3, 20.0);
+        let third = sample(start + 6, 30.0);
+        let mut buffer = AgentBuffer::new(&db, &identity.server_id).await.unwrap();
+        let initial = buffer
+            .receive(&db, &identity, "", vec![first.clone()], true)
+            .await
+            .unwrap();
+        assert_eq!(initial.acknowledgement.unwrap().persisted_through, start);
+
+        let result = buffer
+            .receive(&db, &identity, "", vec![second.clone()], false)
+            .await
+            .unwrap();
+        assert!(result.acknowledgement.is_none());
+        let latest = result.latest.unwrap();
+        let snapshot = HashMap::from([(identity.server_id.clone(), latest)]);
+        assert_eq!(
+            list_servers_with_live(&db, true, &snapshot).await.unwrap()[0].cpu,
+            Some(20.0)
+        );
+        assert_eq!(list_servers(&db, true).await.unwrap()[0].cpu, Some(10.0));
+        let duplicate = buffer
+            .receive(&db, &identity, "", vec![second.clone()], false)
+            .await
+            .unwrap();
+        assert!(duplicate.latest.is_none());
+        let committed = buffer
+            .receive(&db, &identity, "", Vec::new(), true)
+            .await
+            .unwrap();
+        assert_eq!(
+            committed.acknowledgement.unwrap().persisted_through,
+            start + 3
+        );
+        assert!(committed.latest.is_none());
+
+        buffer
+            .receive(&db, &identity, "", vec![third.clone()], false)
+            .await
+            .unwrap();
+        drop(buffer);
+        let mut reconnected = AgentBuffer::new(&db, &identity.server_id).await.unwrap();
+        let replay = reconnected
+            .receive(&db, &identity, "", vec![first, second, third], true)
+            .await
+            .unwrap();
+        let ack = replay.acknowledgement.unwrap();
+        assert_eq!(ack.persisted_through, start + 6);
+        assert_eq!(ack.reports.len(), 1);
+        assert_eq!(
+            history(&db, &identity.server_id, 1)
+                .await
+                .unwrap()
+                .iter()
+                .map(|point| point.sample_count)
+                .sum::<i64>(),
+            3
+        );
+        assert_eq!(
+            ack.reports[0].net_rx_total,
+            replay.latest.unwrap().net_rx_total
+        );
+    }
+
+    #[tokio::test]
+    async fn live_buffer_bounds_pending_samples_and_retains_all_latency_results() {
+        use crate::websocket::ingest::AgentBuffer;
+        let (db, identity) = test_server().await;
+        let start = now() - 2400;
+        let reports: Vec<_> = (0..721)
+            .map(|index| AgentReport {
+                timestamp: start + index * 3,
+                latency_results: vec![AgentLatencyResult {
+                    task_id: "test-task".into(),
+                    timestamp: start + index * 3,
+                    latency_ms: 20.0,
+                    packet_loss: 0.0,
+                }],
+                ..AgentReport::default()
+            })
+            .collect();
+        let mut buffer = AgentBuffer::new(&db, &identity.server_id).await.unwrap();
+        let buffered = buffer
+            .receive(&db, &identity, "", reports[..719].to_vec(), false)
+            .await
+            .unwrap();
+        assert!(buffered.acknowledgement.is_none());
+        assert_eq!(buffered.latest.unwrap().latency_results.len(), 719);
+        let bounded = buffer
+            .receive(&db, &identity, "", reports[719..].to_vec(), false)
+            .await
+            .unwrap();
+        assert_eq!(
+            bounded.acknowledgement.unwrap().persisted_through,
+            reports[718].timestamp
+        );
+        assert_eq!(bounded.latest.unwrap().latency_results.len(), 2);
+        let committed = buffer
+            .receive(&db, &identity, "", Vec::new(), true)
+            .await
+            .unwrap();
+        assert_eq!(
+            committed.acknowledgement.unwrap().persisted_through,
+            reports[720].timestamp
+        );
+        assert_eq!(
+            history(&db, &identity.server_id, 1)
+                .await
+                .unwrap()
+                .iter()
+                .map(|point| point.sample_count)
+                .sum::<i64>(),
+            721
+        );
+    }
+
    #[tokio::test]
    async fn failed_persistence_rolls_back_history_and_watermark() {
        let (db, identity) = test_server().await;
        let start = now() - 60;
-        save_agent_batch(&db, &identity, "initial", &[sample(start, 10.0)], "", true)
+        save_agent_batch(&db, &identity, "initial", &[sample(start, 10.0)], "")
            .await
            .unwrap();
        sqlx::query(
@@ -2753,17 +2954,14 @@ mod tests {
        .await
        .unwrap();
        let next = sample(start + 1, 90.0);
-        assert!(
-            save_agent_batch(
-                &db,
-                &identity,
-                "next",
-                std::slice::from_ref(&next),
-                "",
-                true
-            )
+        let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id)
            .await
-            .is_err()
+            .unwrap();
+        assert!(
+            buffer
+                .receive(&db, &identity, "", vec![next], true)
+                .await
+                .is_err()
        );
        let count = sqlx::query_scalar::<_, i64>(
            "SELECT CAST(SUM(sample_count) AS BIGINT) FROM metric_history",
@@ -2782,10 +2980,11 @@ mod tests {
            .execute(db.pool())
            .await
            .unwrap();
-        let result = save_agent_batch(&db, &identity, "next", &[next], "", true)
+        let result = buffer
+            .receive(&db, &identity, "", Vec::new(), true)
            .await
            .unwrap();
-        assert_eq!(result.persisted_through, start + 1);
+        assert_eq!(result.acknowledgement.unwrap().persisted_through, start + 1);
    }

    #[tokio::test]
@@ -2798,7 +2997,7 @@ mod tests {
            sample(start + 2, 90.0),
            sample(start + 60, 10.0),
        ];
-        save_agent_batch(&db, &identity, "weighted", &reports, "", true)
+        save_agent_batch(&db, &identity, "weighted", &reports, "")
            .await
            .unwrap();
        let points = history(&db, &identity.server_id, 24).await.unwrap();
@@ -2838,7 +3037,7 @@ mod tests {
        let reports = (0..=120)
            .map(|offset| sample(start + offset, if offset == 60 { 10.0 } else { 90.0 }))
            .collect::<Vec<_>>();
-        save_agent_batch(&db, &identity, "alert-data", &reports, "", true)
+        save_agent_batch(&db, &identity, "alert-data", &reports, "")
            .await
            .unwrap();
        for metric in ["cpu", "memory", "disk", "net_in", "net_out"] {
@@ -2894,15 +3093,7 @@ mod tests {
            let identity = identity.clone();
            let reports = reports.clone();
            handles.push(tokio::spawn(async move {
-                save_agent_batch(
-                    &db,
-                    &identity,
-                    &format!("retry-{index}"),
-                    &reports,
-                    "",
-                    true,
-                )
-                .await
+                save_agent_batch(&db, &identity, &format!("retry-{index}"), &reports, "").await
            }));
        }
        for handle in handles {
@@ -3031,7 +3222,7 @@ mod tests {
            })
            .collect::<Vec<_>>();

-        let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1", true)
+        let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1")
            .await
            .unwrap();
        let rows = sqlx::query_scalar::<_, i64>(
@@ -3054,22 +3245,24 @@ mod tests {
        assert_eq!(latency_rows, 600);

        let persisted_at = start + 599;
-        let skipped = save_agent_batch(
-            &db,
-            &identity,
-            "realtime-only",
-            &[AgentReport {
-                timestamp: persisted_at + 1,
-                ..AgentReport::default()
-            }],
-            "127.0.0.1",
-            false,
-        )
-        .await
-        .unwrap();
-        assert!(!skipped.persisted);
-        assert_eq!(skipped.persisted_through, persisted_at);
-        assert_eq!(skipped.next_persist_after_ms, 59_000);
+        let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id)
+            .await
+            .unwrap();
+        let skipped = buffer
+            .receive(
+                &db,
+                &identity,
+                "127.0.0.1",
+                vec![AgentReport {
+                    timestamp: persisted_at + 1,
+                    ..AgentReport::default()
+                }],
+                false,
+            )
+            .await
+            .unwrap();
+        assert!(skipped.acknowledgement.is_none());
+        assert_eq!(skipped.latest.unwrap().timestamp, persisted_at + 1);

        let due = save_agent_batch(
            &db,
@@ -3080,7 +3273,6 @@ mod tests {
                ..AgentReport::default()
            }],
            "127.0.0.1",
-            true,
        )
        .await
        .unwrap();
@@ -3131,7 +3323,6 @@ mod tests {
                ..AgentReport::default()
            }],
            "127.0.0.1",
-            true,
        )
        .await
        .unwrap();
@@ -3291,7 +3482,7 @@ mod tests {
    async fn cleanup_is_bounded_and_keeps_current_data_and_server_state() {
        let (db, identity) = test_server().await;
        let current = now();
-        save_agent_batch(&db, &identity, "latest", &[sample(current, 40.0)], "", true)
+        save_agent_batch(&db, &identity, "latest", &[sample(current, 40.0)], "")
            .await
            .unwrap();
        let task_id = create_latency_task(
@@ -24,12 +24,9 @@ use clap::Parser;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::sync::atomic::Ordering;
 use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, watch};
 use tower_http::trace::TraceLayer;

-const LIVE_REPORT_DIVISOR: i64 = 15;
-
 pub struct AppState {
    pub db: db::Database,
    pub config: config::Config,
@@ -38,6 +35,7 @@ pub struct AppState {
    pub http: reqwest::Client,
    pub agents: RwLock<HashMap<String, websocket::AgentConnection>>,
    pub last_agent_reports: RwLock<HashMap<String, i64>>,
+    pub live_reports: RwLock<HashMap<String, models::AgentReport>>,
    pub started_at: i64,
    pub dashboard_tx: broadcast::Sender<websocket::DashboardEvent>,
    pub database_maintenance: Mutex<()>,
@@ -47,13 +45,13 @@ pub struct AppState {
    pub restart_tx: watch::Sender<bool>,
    pub login_attempts: security::AttemptLimiter,
    pub sensitive_attempts: security::AttemptLimiter,
-    pub wake_requests: security::IntervalLimiter,
    pub password_verifications: Arc<Semaphore>,
 }

 impl AppState {
    pub async fn disconnect_agents(&self) {
        let agents = std::mem::take(&mut *self.agents.write().await);
+        self.live_reports.write().await.clear();
        for connection in agents.into_values() {
            let _ = connection.sender.try_send(websocket::AgentCommand::Close);
        }
@@ -84,52 +82,12 @@ impl AppState {

    pub async fn disconnect_agent(&self, server_id: &str) {
        let connection = self.agents.write().await.remove(server_id);
+        self.live_reports.write().await.remove(server_id);
        if let Some(connection) = connection {
            let _ = connection.sender.try_send(websocket::AgentCommand::Close);
        }
    }

-    pub async fn wake_agent(&self, server_id: &str) {
-        let Some(connection) = self.agents.read().await.get(server_id).cloned() else {
-            return;
-        };
-        connection
-            .live_until
-            .store(db::now().saturating_add(75), Ordering::Release);
-        let payload = serde_json::json!({
-            "type": "ack",
-            "ts": db::now(),
-            "persisted": false,
-            "persistenceError": false,
-            "persistedThroughTs": 0,
-            "nextPersistAfterMs": connection.report_interval.clamp(15, 3600) * 1000,
-            "nextWssReportAfterMs": connection.collect_interval.clamp(1, 60) * 1000,
-            "realtimeHint": true,
-        })
-        .to_string();
-        let _ = connection
-            .sender
-            .try_send(websocket::AgentCommand::Text(payload));
-    }
-
-    pub async fn agent_wss_interval_ms(&self, server_id: &str) -> u64 {
-        let Some(connection) = self.agents.read().await.get(server_id).cloned() else {
-            return 60_000;
-        };
-        let seconds = if connection.live_until.load(Ordering::Acquire) >= db::now() {
-            connection.collect_interval.clamp(1, 60)
-        } else {
-            (connection
-                .report_interval
-                .clamp(15, 3600)
-                .saturating_add(LIVE_REPORT_DIVISOR - 1)
-                / LIVE_REPORT_DIVISOR)
-                .clamp(1, 60)
-                .max(connection.collect_interval.clamp(1, 60))
-        };
-        seconds as u64 * 1000
-    }
-
    pub async fn send_remote_task(
        &self,
        task: &models::RemoteTaskInfo,
@@ -210,6 +168,7 @@ async fn main() -> Result<()> {
        http,
        agents: RwLock::new(HashMap::new()),
        last_agent_reports: RwLock::new(HashMap::new()),
+        live_reports: RwLock::new(HashMap::new()),
        started_at: db::now(),
        dashboard_tx,
        database_maintenance: Mutex::new(()),
@@ -227,7 +186,6 @@ async fn main() -> Result<()> {
            std::time::Duration::from_secs(5 * 60),
            std::time::Duration::from_secs(10 * 60),
        ),
-        wake_requests: security::IntervalLimiter::new(std::time::Duration::from_secs(1)),
        password_verifications: Arc::new(Semaphore::new(4)),
    });

@@ -350,7 +308,6 @@ async fn main() -> Result<()> {
        .route("/api/history/{id}", get(routes::public::history))
        .route("/api/latency/{id}", get(routes::public::latency_history))
        .route("/api/exchange-rates", get(routes::public::exchange_rates))
-        .route("/api/live/wake", post(routes::public::wake_servers))
        .route(
            "/api/turnstile/verify",
            post(routes::auth::verify_turnstile),
@@ -1,3 +1,7 @@
+pub use nodeflare_telemetry::{
+    DiskMetric as AgentDiskMetric, GpuMetric as AgentGpuMetric,
+    LatencyResult as AgentLatencyResult, Report as AgentReport,
+};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;

@@ -139,83 +143,6 @@ pub struct LoginSessionView {
    pub current: bool,
 }

-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct AgentReport {
-    pub timestamp: i64,
-    pub cpu: f64,
-    pub load1: f64,
-    pub load5: f64,
-    pub load15: f64,
-    pub mem_used: i64,
-    pub mem_total: i64,
-    pub swap_used: i64,
-    pub swap_total: i64,
-    pub disk_used: i64,
-    pub disk_total: i64,
-    pub net_in: f64,
-    pub net_out: f64,
-    pub net_rx_total: i64,
-    pub net_tx_total: i64,
-    pub uptime: i64,
-    pub processes: i64,
-    pub tcp_connections: i64,
-    pub udp_connections: i64,
-    pub cpu_cores: i64,
-    pub cpu_model: String,
-    pub os: String,
-    pub kernel: String,
-    pub arch: String,
-    pub virtualization: String,
-    pub gpu_usage: f64,
-    pub gpu_model: String,
-    pub agent_version: String,
-    pub ip_v4: String,
-    pub ip_v6: String,
-    pub disk_read_bps: f64,
-    pub disk_write_bps: f64,
-    pub disk_read_iops: f64,
-    pub disk_write_iops: f64,
-    pub disk_await_ms: f64,
-    pub disk_utilization: f64,
-    pub disks: Vec<AgentDiskMetric>,
-    pub gpus: Vec<AgentGpuMetric>,
-    pub latency_results: Vec<AgentLatencyResult>,
-}
-
-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct AgentDiskMetric {
-    pub name: String,
-    pub mount_point: String,
-    pub used: i64,
-    pub total: i64,
-    pub read_bps: f64,
-    pub write_bps: f64,
-    pub read_iops: f64,
-    pub write_iops: f64,
-    pub await_ms: f64,
-    pub utilization: f64,
-}
-
-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct AgentGpuMetric {
-    pub model: String,
-    pub usage: Option<f64>,
-    pub memory_used: i64,
-    pub memory_total: i64,
-}
-
-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct AgentLatencyResult {
-    pub task_id: String,
-    pub timestamp: i64,
-    pub latency_ms: f64,
-    pub packet_loss: f64,
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub struct ServerView {
    pub id: String,
@@ -447,12 +374,6 @@ pub struct ServerBatchInput {
    pub ids: Vec<String>,
 }

-#[derive(Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct WakeServersInput {
-    pub server_ids: Vec<String>,
-}
-
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct Enable2FaRequest {
@@ -34,7 +34,8 @@ pub struct DatabaseMigrationInput {
 }

 pub async fn servers_get(State(state): State<Arc<AppState>>) -> Result<Response, ApiResponse> {
-    let servers = crate::db::queries::list_servers(&state.db, true)
+    let live = state.live_reports.read().await.clone();
+    let servers = crate::db::queries::list_servers_with_live(&state.db, true, &live)
        .await
        .map_err(ApiResponse::internal)?;
    Ok(Json(serde_json::json!({"servers": servers})).into_response())
@@ -871,7 +872,8 @@ fn validate_server(input: &ServerInput) -> Result<(), &'static str> {
    }
    if !(1..=31).contains(&input.reset_day)
        || !(15..=3600).contains(&input.report_interval)
-        || !(1..=60).contains(&input.collect_interval)
+        || !(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64..=60)
+            .contains(&input.collect_interval)
        || input.collect_interval > input.report_interval
        || (input.report_interval + input.collect_interval - 1) / input.collect_interval > 720
    {
@@ -1125,7 +1127,7 @@ mod server_validation_tests {
            "hidden": false, "expires_at": null, "traffic_limit": 0,
            "traffic_limit_type": "sum", "price": 0, "billing_cycle": 30,
            "currency": "CNY", "auto_renewal": false, "network_interface": "",
-            "reset_day": 1, "report_interval": 60, "collect_interval": 1,
+            "reset_day": 1, "report_interval": 60, "collect_interval": 3,
            "rx_correction": 0, "tx_correction": 0, "agent_mirror": "",
            "offline_notify_disabled": false, "auto_update": true
        }))
@@ -1,14 +1,13 @@
-use super::{ApiResponse, bearer_or_cookie, client_ip, cookie};
+use super::{ApiResponse, bearer_or_cookie, cookie};
 use crate::AppState;
 use crate::db::Settings;
-use crate::models::{PublicConfig, WakeServersInput};
+use crate::models::PublicConfig;
 use axum::Json;
-use axum::extract::{ConnectInfo, Path, Query, State};
-use axum::http::{HeaderMap, StatusCode};
+use axum::extract::{Path, Query, State};
+use axum::http::HeaderMap;
 use axum::response::{IntoResponse, Response};
 use serde::Serialize;
 use std::collections::HashMap;
-use std::net::SocketAddr;
 use std::sync::Arc;

 #[derive(Serialize)]
@@ -98,7 +97,8 @@ pub async fn bootstrap(
            exchange_rates: None,
        }));
    }
-    let servers = crate::db::queries::list_servers(&state.db, false)
+    let live = state.live_reports.read().await.clone();
+    let servers = crate::db::queries::list_servers_with_live(&state.db, false, &live)
        .await
        .map_err(ApiResponse::internal)?
        .into_iter()
@@ -167,26 +167,6 @@ pub async fn exchange_rates(State(state): State<Arc<AppState>>) -> Result<Respon
    .into_response())
 }

-pub async fn wake_servers(
-    State(state): State<Arc<AppState>>,
-    ConnectInfo(peer): ConnectInfo<SocketAddr>,
-    headers: HeaderMap,
-    Json(input): Json<WakeServersInput>,
-) -> Result<Response, ApiResponse> {
-    require_dashboard(&state, &headers).await?;
-    if input.server_ids.len() > 500 {
-        return Err(ApiResponse::bad_request("服务器列表过长"));
-    }
-    let client = client_ip(&headers, peer, &state.config.trusted_proxies);
-    if !state.wake_requests.allow(&client) {
-        return Ok(StatusCode::NO_CONTENT.into_response());
-    }
-    for server_id in input.server_ids {
-        state.wake_agent(&server_id).await;
-    }
-    Ok(StatusCode::NO_CONTENT.into_response())
-}
-
 async fn require_dashboard(state: &AppState, headers: &HeaderMap) -> Result<(), ApiResponse> {
    let settings = crate::db::load_settings(&state.db)
        .await
@@ -18,38 +18,6 @@ pub struct AttemptLimiter {
    retention: Duration,
 }

-pub struct IntervalLimiter {
-    entries: Mutex<HashMap<String, Instant>>,
-    minimum_interval: Duration,
-    retention: Duration,
-}
-
-impl IntervalLimiter {
-    pub fn new(minimum_interval: Duration) -> Self {
-        Self {
-            entries: Mutex::new(HashMap::new()),
-            minimum_interval,
-            retention: minimum_interval.saturating_mul(3_600),
-        }
-    }
-
-    pub fn allow(&self, key: &str) -> bool {
-        let now = Instant::now();
-        let Ok(mut entries) = self.entries.lock() else {
-            return false;
-        };
-        entries.retain(|_, seen| now.duration_since(*seen) <= self.retention);
-        if entries
-            .get(key)
-            .is_some_and(|seen| now.duration_since(*seen) < self.minimum_interval)
-        {
-            return false;
-        }
-        entries.insert(key.to_string(), now);
-        true
-    }
-}
-
 impl AttemptLimiter {
    pub fn new(maximum_failures: u32, window: Duration, block_for: Duration) -> Self {
        Self {
@@ -123,12 +91,4 @@ mod tests {
        limiter.clear("client");
        assert_eq!(limiter.retry_after("client"), None);
    }
-
-    #[test]
-    fn interval_limiter_coalesces_repeated_requests() {
-        let limiter = IntervalLimiter::new(Duration::from_secs(60));
-        assert!(limiter.allow("client"));
-        assert!(!limiter.allow("client"));
-        assert!(limiter.allow("other"));
-    }
 }
@@ -8,26 +8,15 @@ use axum::extract::{ConnectInfo, State};
 use axum::http::HeaderMap;
 use axum::response::{IntoResponse, Response};
 use futures_util::{SinkExt, StreamExt};
+use nodeflare_telemetry as telemetry;
 use serde::Deserialize;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::sync::atomic::AtomicI64;
 use tokio::sync::mpsc;

 const MAX_AGENT_MESSAGE_BYTES: usize = 2 * 1024 * 1024;
 const MAX_TASK_RESULT_BYTES: usize = 1024 * 1024;

-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct UpdateMessage {
-    #[serde(rename = "type")]
-    _message_type: String,
-    #[serde(rename = "batchId")]
-    batch_id: String,
-    samples: Vec<AgentReport>,
-    persist: bool,
-}
-
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 struct TaskResultMessage {
@@ -92,14 +81,12 @@ async fn run(
        return;
    };
    let (outbound_tx, mut outbound_rx) = mpsc::channel::<AgentCommand>(256);
+    let mut info = None;
    let previous = state.agents.write().await.insert(
        identity.server_id.clone(),
        AgentConnection {
            connection_id: connection_id.clone(),
            sender: outbound_tx.clone(),
-            report_interval: identity.report_interval,
-            collect_interval: identity.collect_interval,
-            live_until: Arc::new(AtomicI64::new(0)),
        },
    );
    if let Some(previous) = previous {
@@ -107,11 +94,12 @@ async fn run(
    }

    // Recheck the token after registration in case the node changed meanwhile.
+    let initialized = super::ingest::AgentBuffer::new(&state.db, &identity.server_id).await;
    let still_authorized = matches!(
        crate::db::queries::agent_identity(&state.db, &token).await,
        Ok(Some(current)) if current.server_id == identity.server_id
    );
-    if !still_authorized {
+    if !still_authorized || initialized.is_err() {
        let mut agents = state.agents.write().await;
        if agents
            .get(&identity.server_id)
@@ -124,6 +112,7 @@ async fn run(
        tracing::warn!(server_id = %identity.server_id, "rejected stale Agent connection");
        return;
    }
+    let mut buffer = initialized.expect("Agent buffer was initialized");
    tracing::info!(server_id = %identity.server_id, remote_ip, "Agent connected");

    if let Some(config) = crate::db::queries::agent_config(&state.db, &identity.server_id)
@@ -166,9 +155,80 @@ async fn run(
    });

    while let Some(message) = websocket_rx.next().await {
+        // Registration waits for in-flight ingestion before loading the next connection's watermark.
+        let connections = state.agents.read().await;
+        if !connections
+            .get(&identity.server_id)
+            .is_some_and(|connection| connection.connection_id == connection_id)
+        {
+            break;
+        }
        match message {
+            Ok(Message::Binary(bytes)) => {
+                let result = async {
+                    let _activity = state
+                        .database_activity
+                        .write()
+                        .map_err(|_| anyhow::anyhow!("database is under maintenance"))?;
+                    let update: telemetry::Update = telemetry::decode(&bytes)?;
+                    let persist = update.persist;
+                    let reports = update.into_reports(&mut info)?;
+                    let result = buffer
+                        .receive(&state.db, &identity, &remote_ip, reports, persist)
+                        .await?;
+                    state
+                        .last_agent_reports
+                        .write()
+                        .await
+                        .insert(identity.server_id.clone(), crate::db::now());
+                    if let Some(acknowledgement) = result.acknowledgement {
+                        send_ack(&outbound_tx, &acknowledgement);
+                        if !acknowledgement.reports.is_empty() {
+                            evaluate_alerts(&state, &identity.server_id);
+                        }
+                    }
+                    if let Some(report) = result.latest {
+                        let mut live = state.live_reports.write().await;
+                        if live
+                            .get(&identity.server_id)
+                            .is_none_or(|current| current.timestamp < report.timestamp)
+                        {
+                            broadcast_report(&state, &identity, &report);
+                            let mut cached = report;
+                            if let Some(previous) = live.get(&identity.server_id) {
+                                cached
+                                    .latency_results
+                                    .extend(previous.latency_results.iter().cloned());
+                            }
+                            cached.latency_results.sort_by(|left, right| {
+                                left.task_id
+                                    .cmp(&right.task_id)
+                                    .then(right.timestamp.cmp(&left.timestamp))
+                            });
+                            cached
+                                .latency_results
+                                .dedup_by(|left, right| left.task_id == right.task_id);
+                            cached
+                                .latency_results
+                                .retain(|value| value.timestamp >= crate::db::now() - 7200);
+                            cached
+                                .latency_results
+                                .sort_by_key(|value| std::cmp::Reverse(value.timestamp));
+                            cached.latency_results.truncate(4096);
+                            live.insert(identity.server_id.clone(), cached);
+                        }
+                    }
+                    Ok::<_, anyhow::Error>(())
+                }
+                .await;
+                if let Err(error) = result {
+                    tracing::warn!(%error, server_id = %identity.server_id, "Agent telemetry failed");
+                    send_persistence_error(&outbound_tx, 0);
+                    break;
+                }
+            }
            Ok(Message::Text(text)) if text.len() <= MAX_AGENT_MESSAGE_BYTES => {
-                handle_text(&state, &identity, &remote_ip, &outbound_tx, &text).await;
+                handle_text(&state, &identity, &outbound_tx, &text).await;
            }
            Ok(Message::Ping(payload)) => {
                let _ = outbound_tx.try_send(AgentCommand::Pong(payload.to_vec()));
@@ -193,7 +253,6 @@ async fn run(
 async fn handle_text(
    state: &Arc<AppState>,
    identity: &AgentIdentity,
-    remote_ip: &str,
    outbound: &mpsc::Sender<AgentCommand>,
    text: &str,
 ) {
@@ -205,80 +264,6 @@ async fn handle_text(
        return;
    };
    match value.get("type").and_then(serde_json::Value::as_str) {
-        Some("update") => {
-            let Ok(update) = serde_json::from_value::<UpdateMessage>(value) else {
-                send_persistence_error(outbound, 0);
-                return;
-            };
-            if !valid_batch_id(&update.batch_id) {
-                send_persistence_error(outbound, 0);
-                return;
-            }
-            match crate::db::queries::save_agent_batch(
-                &state.db,
-                identity,
-                &update.batch_id,
-                &update.samples,
-                remote_ip,
-                update.persist,
-            )
-            .await
-            {
-                Ok(result) => {
-                    state
-                        .last_agent_reports
-                        .write()
-                        .await
-                        .insert(identity.server_id.clone(), crate::db::now());
-                    let next_wss_report_after_ms =
-                        state.agent_wss_interval_ms(&identity.server_id).await;
-                    send_ack(outbound, &result, next_wss_report_after_ms);
-                    if !result.reports.is_empty() {
-                        broadcast_reports(state, identity, &result.reports);
-                    }
-                    if result.persisted && !result.reports.is_empty() {
-                        let state = Arc::clone(state);
-                        let server_id = identity.server_id.clone();
-                        tokio::spawn(async move {
-                            let Ok(_activity) = state.database_activity.write() else {
-                                return;
-                            };
-                            let settings = match crate::db::load_settings(&state.db).await {
-                                Ok(settings) => settings,
-                                Err(error) => {
-                                    tracing::error!(%error, "failed to load alert settings");
-                                    return;
-                                }
-                            };
-                            if !settings.notification_enabled {
-                                return;
-                            }
-                            let server_name =
-                                crate::db::queries::server_name(&state.db, &server_id)
-                                    .await
-                                    .ok()
-                                    .flatten()
-                                    .unwrap_or_else(|| server_id.clone());
-                            if let Err(error) = crate::notify::evaluate_resource_alerts(
-                                &state.db,
-                                &settings,
-                                &server_id,
-                                &server_name,
-                            )
-                            .await
-                            {
-                                tracing::error!(%error, server_id, "resource alert evaluation failed");
-                            }
-                            state.notification_wake.notify_one();
-                        });
-                    }
-                }
-                Err(error) => {
-                    tracing::error!(%error, server_id = %identity.server_id, "Agent batch persistence failed");
-                    send_persistence_error(outbound, 0);
-                }
-            }
-        }
        Some("task_result") => {
            let Ok(result) = serde_json::from_value::<TaskResultMessage>(value) else {
                return;
@@ -335,11 +320,39 @@ async fn handle_text(
    }
 }

-fn send_ack(
-    outbound: &mpsc::Sender<AgentCommand>,
-    result: &PersistResult,
-    next_wss_report_after_ms: u64,
-) {
+fn evaluate_alerts(state: &Arc<AppState>, server_id: &str) {
+    let state = Arc::clone(state);
+    let server_id = server_id.to_string();
+    tokio::spawn(async move {
+        let Ok(_activity) = state.database_activity.write() else {
+            return;
+        };
+        let settings = match crate::db::load_settings(&state.db).await {
+            Ok(settings) => settings,
+            Err(error) => {
+                tracing::error!(%error, "failed to load alert settings");
+                return;
+            }
+        };
+        if !settings.notification_enabled {
+            return;
+        }
+        let server_name = crate::db::queries::server_name(&state.db, &server_id)
+            .await
+            .ok()
+            .flatten()
+            .unwrap_or_else(|| server_id.clone());
+        if let Err(error) =
+            crate::notify::evaluate_resource_alerts(&state.db, &settings, &server_id, &server_name)
+                .await
+        {
+            tracing::error!(%error, server_id, "resource alert evaluation failed");
+        }
+        state.notification_wake.notify_one();
+    });
+}
+
+fn send_ack(outbound: &mpsc::Sender<AgentCommand>, result: &PersistResult) {
    let _ = outbound.try_send(AgentCommand::Text(
        serde_json::json!({
            "type": "ack",
@@ -348,8 +361,6 @@ fn send_ack(
            "persistenceError": false,
            "persistedThroughTs": result.persisted_through,
            "nextPersistAfterMs": result.next_persist_after_ms,
-            "nextWssReportAfterMs": next_wss_report_after_ms,
-            "realtimeHint": false,
        })
        .to_string(),
    ));
@@ -364,58 +375,43 @@ fn send_persistence_error(outbound: &mpsc::Sender<AgentCommand>, persisted: i64)
            "persistenceError": true,
            "persistedThroughTs": persisted,
            "nextPersistAfterMs": 5000,
-            "nextWssReportAfterMs": 5000,
-            "realtimeHint": false,
        })
        .to_string(),
    ));
 }

-fn broadcast_reports(state: &AppState, identity: &AgentIdentity, reports: &[AgentReport]) {
-    if identity.hidden {
+fn broadcast_report(state: &AppState, identity: &AgentIdentity, report: &AgentReport) {
+    if identity.hidden || state.dashboard_tx.receiver_count() == 0 {
        return;
    }
-    let samples = reports
-        .iter()
-        .map(|report| {
-            let mut data = serde_json::to_value(report).unwrap_or_else(|_| serde_json::json!({}));
-            if let Some(object) = data.as_object_mut() {
-                for key in [
-                    "timestamp",
-                    "cpu_model",
-                    "os",
-                    "kernel",
-                    "arch",
-                    "virtualization",
-                    "gpu_model",
-                    "agent_version",
-                    "ip_v4",
-                    "ip_v6",
-                ] {
-                    object.remove(key);
-                }
-            }
-            serde_json::json!({"ts": report.timestamp, "data": data})
-        })
-        .collect::<Vec<_>>();
-    let payload = serde_json::json!({
+    let mut data = serde_json::to_value(report).unwrap_or_else(|_| serde_json::json!({}));
+    if let Some(object) = data.as_object_mut() {
+        for key in [
+            "timestamp",
+            "cpu_model",
+            "os",
+            "kernel",
+            "arch",
+            "virtualization",
+            "gpu_model",
+            "agent_version",
+            "ip_v4",
+            "ip_v6",
+        ] {
+            object.remove(key);
+        }
+    }
+    let payload = telemetry::encode(&serde_json::json!({
        "type": "batchUpdate",
        "ts": crate::db::now(),
-        "updates": [{"serverId": identity.server_id, "samples": samples}],
-    })
-    .to_string();
-    let _ = state.dashboard_tx.send(DashboardEvent {
-        server_id: identity.server_id.clone(),
-        payload,
-    });
-}
-
-fn valid_batch_id(value: &str) -> bool {
-    !value.is_empty()
-        && value.len() <= 128
-        && value
-            .bytes()
-            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
+        "updates": [{"serverId": identity.server_id, "samples": [{"ts": report.timestamp, "data": data}]}],
+    }));
+    if let Ok(payload) = payload {
+        let _ = state.dashboard_tx.send(DashboardEvent {
+            server_id: identity.server_id.clone(),
+            payload: payload.into(),
+        });
+    }
 }

 fn valid_task_id(value: &str) -> bool {
@@ -452,9 +448,6 @@ mod tests {
        let connection = AgentConnection {
            connection_id: "test".to_string(),
            sender,
-            report_interval: 60,
-            collect_interval: 1,
-            live_until: Arc::new(AtomicI64::new(0)),
        };
        let task = RemoteTaskInfo {
            id: uuid::Uuid::new_v4().to_string(),
@@ -493,14 +486,14 @@ mod tests {

    #[test]
    fn updates_require_an_explicit_persistence_flag() {
-        let mut message = serde_json::json!({"type": "update", "batchId": "batch", "samples": []});
-        assert!(serde_json::from_value::<UpdateMessage>(message.clone()).is_err());
+        let mut message = serde_json::json!({"samples": []});
+        assert!(serde_json::from_value::<telemetry::Update>(message.clone()).is_err());
        message["persist"] = serde_json::Value::Null;
-        assert!(serde_json::from_value::<UpdateMessage>(message.clone()).is_err());
+        assert!(serde_json::from_value::<telemetry::Update>(message.clone()).is_err());
        for persist in [true, false] {
            message["persist"] = persist.into();
            assert_eq!(
-                serde_json::from_value::<UpdateMessage>(message.clone())
+                serde_json::from_value::<telemetry::Update>(message.clone())
                    .unwrap()
                    .persist,
                persist
@@ -86,11 +86,15 @@ async fn run(
        tokio::select! {
            update = updates.recv() => match update {
                Ok(update) if server_id.as_ref().is_none_or(|id| id == &update.server_id) => {
-                    if sender.send(Message::Text(update.payload.into())).await.is_err() {
+                    if sender.send(Message::Binary(update.payload)).await.is_err() {
                        break;
                    }
                }
-                Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
+                Ok(_) => {}
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
+                    let _ = sender.send(Message::Close(None)).await;
+                    break;
+                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            },
            incoming = receiver.next() => match incoming {
@@ -0,0 +1,110 @@
+use crate::db::{
+    Database, now,
+    queries::{self, AgentIdentity, PersistResult, TrafficState},
+};
+use crate::models::AgentReport;
+use anyhow::{Result, ensure};
+use nodeflare_telemetry::{MAX_BATCH_BYTES, MAX_SAMPLES};
+
+pub struct AgentBuffer {
+    pending: Vec<AgentReport>,
+    pending_bytes: usize,
+    received_through: i64,
+    persisted_through: i64,
+    traffic: TrafficState,
+}
+
+pub struct IngestResult {
+    pub latest: Option<AgentReport>,
+    pub acknowledgement: Option<PersistResult>,
+}
+
+impl AgentBuffer {
+    pub async fn new(db: &Database, server_id: &str) -> Result<Self> {
+        let traffic = queries::agent_traffic_state(db, server_id).await?;
+        Ok(Self {
+            pending: Vec::new(),
+            pending_bytes: 0,
+            received_through: traffic.timestamp,
+            persisted_through: traffic.timestamp,
+            traffic,
+        })
+    }
+
+    pub async fn receive(
+        &mut self,
+        db: &Database,
+        identity: &AgentIdentity,
+        remote_ip: &str,
+        mut reports: Vec<AgentReport>,
+        persist: bool,
+    ) -> Result<IngestResult> {
+        let current = now();
+        ensure!(reports.len() <= MAX_SAMPLES, "too many telemetry samples");
+        ensure!(
+            reports.iter().all(|report| report.timestamp > 0
+                && report.timestamp >= current - 7200
+                && report.timestamp <= current + 300
+                && queries::valid_agent_report(report, current)),
+            "invalid telemetry sample"
+        );
+        reports.sort_by_key(|report| report.timestamp);
+        reports.dedup_by_key(|report| report.timestamp);
+        reports.retain(|report| report.timestamp > self.received_through);
+        let bytes = reports.iter().try_fold(0_usize, |size, report| {
+            serde_json::to_vec(report).map(|encoded| size.saturating_add(encoded.len()))
+        })?;
+        ensure!(bytes <= MAX_BATCH_BYTES, "telemetry batch is too large");
+        let mut acknowledgement = None;
+        if !self.pending.is_empty()
+            && (self.pending.len() + reports.len() > MAX_SAMPLES
+                || self.pending_bytes.saturating_add(bytes) > MAX_BATCH_BYTES)
+        {
+            acknowledgement = Some(self.flush(db, identity, remote_ip).await?);
+        }
+        let mut latest = None;
+        let mut latency = Vec::new();
+        for report in &reports {
+            let mut live = report.clone();
+            queries::apply_traffic(&mut live, &mut self.traffic, identity);
+            self.received_through = live.timestamp;
+            latency.append(&mut live.latency_results);
+            latest = Some(live);
+        }
+        if let Some(latest) = &mut latest {
+            latest.latency_results = latency;
+        }
+        self.pending.extend(reports);
+        self.pending_bytes = self.pending_bytes.saturating_add(bytes);
+        if persist || self.pending_bytes >= MAX_BATCH_BYTES || self.pending.len() >= MAX_SAMPLES {
+            acknowledgement = Some(self.flush(db, identity, remote_ip).await?);
+        }
+        Ok(IngestResult {
+            latest,
+            acknowledgement,
+        })
+    }
+
+    async fn flush(
+        &mut self,
+        db: &Database,
+        identity: &AgentIdentity,
+        remote_ip: &str,
+    ) -> Result<PersistResult> {
+        let Some(first) = self.pending.first() else {
+            return Ok(PersistResult {
+                reports: Vec::new(),
+                persisted: true,
+                persisted_through: self.persisted_through,
+                next_persist_after_ms: identity.report_interval.clamp(15, 3600) as u64 * 1000,
+            });
+        };
+        let batch_id = format!("{}:{}", first.timestamp, self.received_through);
+        let result =
+            queries::save_agent_batch(db, identity, &batch_id, &self.pending, remote_ip).await?;
+        self.persisted_through = result.persisted_through;
+        self.pending.clear();
+        self.pending_bytes = 0;
+        Ok(result)
+    }
+}
@@ -1,17 +1,13 @@
 pub mod agent;
 pub mod dashboard;
+pub(crate) mod ingest;

-use std::sync::Arc;
-use std::sync::atomic::AtomicI64;
 use tokio::sync::mpsc;

 #[derive(Clone)]
 pub struct AgentConnection {
    pub connection_id: String,
    pub sender: mpsc::Sender<AgentCommand>,
-    pub report_interval: i64,
-    pub collect_interval: i64,
-    pub live_until: Arc<AtomicI64>,
 }

 #[derive(Debug)]
@@ -24,5 +20,5 @@ pub enum AgentCommand {
 #[derive(Debug, Clone)]
 pub struct DashboardEvent {
    pub server_id: String,
-    pub payload: String,
+    pub payload: axum::body::Bytes,
 }
@@ -96,7 +96,7 @@ function VerificationGate() {
 }

 function HomeView() {
-  const { carrierLatency, config, error, exchangeRates, liveMetrics, loading, openServer, reload, servers } = useApp();
+  const { carrierLatency, config, error, exchangeRates, liveConnected, liveMetrics, loading, openServer, reload, servers } = useApp();
  const [query, setQuery] = useState("");
  const [group, setGroup] = useState("__all__");
  const locale = config.locale;
@@ -125,7 +125,7 @@ function HomeView() {
        <div className="dashboard-loading"><span className="loading-ring" aria-hidden="true" /><p>{ui(locale, "加载中…", "Loading…")}</p></div>
      ) : visible.length ? (
        <section className={`node-grid ${carrierLatency ? "carrier-latency" : ""}`}>
-          {visible.map((server) => <NodeCard key={server.id} server={server} config={config} liveLatencyResults={liveMetrics[server.id]?.latencyResults} onOpen={() => openServer(server)} />)}
+          {visible.map((server) => <NodeCard key={server.id} server={server} config={config} liveConnected={liveConnected} liveLatencyResults={liveMetrics[server.id]?.latencyResults} onOpen={() => openServer(server)} />)}
        </section>
      ) : !error ? (
        <div className="empty-state"><strong>{servers.length ? ui(locale, "没有匹配的节点", "No matching servers") : ui(locale, "尚未添加节点", "No servers added")}</strong></div>
@@ -13,7 +13,6 @@ import { useFavicon, useStoredAppearance, useSystemDark } from "./hooks/useBrows
 import { resolveBackground, themeToggle } from "./theme";
 import {
  BOOTSTRAP_POLL_INTERVAL_MS,
-  createLiveFlushScheduler,
  createRefreshQueue,
  shouldSyncBootstrap,
 } from "./refresh";
@@ -35,6 +34,7 @@ export interface AppState {
  configReady: boolean;
  servers: Server[];
  liveMetrics: LiveMetricsMap;
+  liveConnected: boolean;
  exchangeRates: ExchangeRates | null;
  loading: boolean;
  error: string;
@@ -72,6 +72,8 @@ export function AppProvider({ children }: { children: ReactNode }) {
  const [configReady, setConfigReady] = useState(demoMode);
  const [servers, setServers] = useState<Server[]>([]);
  const [liveMetrics, setLiveMetrics] = useState<LiveMetricsMap>({});
+  const [liveConnected, setLiveConnected] = useState(false);
+  const pendingLive = useRef<BatchUpdate[]>([]);
  const [clockNow, setClockNow] = useState(() => Date.now());
  const [exchangeRates, setExchangeRates] = useState<ExchangeRates | null>(null);
  const [loading, setLoading] = useState(true);
@@ -217,12 +219,19 @@ export function AppProvider({ children }: { children: ReactNode }) {

  useEffect(() => {
    let timer: number | undefined;
+    const tick = () => {
+      setClockNow(Date.now());
+      if (!pendingLive.current.length) return;
+      const batch = pendingLive.current;
+      pendingLive.current = [];
+      setLiveMetrics((current) => applyBatch(current, batch, serversRef.current));
+    };
    const sync = () => {
      window.clearInterval(timer);
      timer = undefined;
      if (document.hidden || navigator.onLine === false) return;
-      setClockNow(Date.now());
-      timer = window.setInterval(() => setClockNow(Date.now()), 1_000);
+      tick();
+      timer = window.setInterval(tick, 1_000);
    };
    sync();
    document.addEventListener("visibilitychange", sync);
@@ -239,43 +248,24 @@ export function AppProvider({ children }: { children: ReactNode }) {
  useEffect(() => {
    if (!configReady || access !== "ok" || demoMode) return;
    let active = true;
-    let pending: BatchUpdate[] = [];
-    // Coalesce independently timed reports onto a shared one-second UI tick.
-    const flush = () => {
-      if (!pending.length) return;
-      const batch = pending;
-      pending = [];
-      setLiveMetrics((current) => applyBatch(current, batch, serversRef.current));
-    };
-    const scheduler = createLiveFlushScheduler(flush);
    const disconnect = connectLive({ serverId: selectedId }, {
      onBatch: (updates) => {
-        pending.push(...updates);
-        scheduler.schedule();
+        pendingLive.current.push(...updates);
      },
      onConnectedChange: (connected) => {
        liveConnectedRef.current = connected;
+        setLiveConnected(connected);
        if (!connected) {
-          scheduler.cancel();
-          pending = [];
+          pendingLive.current = [];
        }
        if (active && !connected && !document.hidden && navigator.onLine !== false) {
          void reload(true);
        }
      },
-      onWakeRequested: async () => {
-        const serverIds = selectedId ? [selectedId] : serversRef.current.map((server) => server.id);
-        for (let offset = 0; offset < serverIds.length; offset += 500) {
-          // The wake endpoint accepts one request per second and at most 500 nodes.
-          if (offset > 0) await new Promise((resolve) => window.setTimeout(resolve, 1_100));
-          if (!active || document.hidden || navigator.onLine === false) return;
-          await api.wakeServers(serverIds.slice(offset, offset + 500));
-        }
-      },
    });
    return () => {
      active = false;
-      scheduler.cancel();
+      pendingLive.current = [];
      disconnect();
    };
  }, [access, configReady, reload, selectedId]);
@@ -333,6 +323,7 @@ export function AppProvider({ children }: { children: ReactNode }) {
    configReady,
    servers: liveServers,
    liveMetrics,
+    liveConnected,
    exchangeRates,
    loading,
    error,
@@ -349,7 +340,7 @@ export function AppProvider({ children }: { children: ReactNode }) {
    reload: () => reload(),
    login,
    verify,
-  }), [access, background, blur, carrierLatency, config, configReady, dark, error, exchangeRates, goHome, liveMetrics, liveServers, loading, login, openServer, reload, selectedId, toggleTheme, verify]);
+  }), [access, background, blur, carrierLatency, config, configReady, dark, error, exchangeRates, goHome, liveConnected, liveMetrics, liveServers, loading, login, openServer, reload, selectedId, toggleTheme, verify]);

  return <AppContext.Provider value={value}>{children}</AppContext.Provider>;
 }
@@ -45,11 +45,6 @@ export const api = {
  bootstrap: () => request<Bootstrap>("/api/bootstrap", { cache: "no-store", signal: AbortSignal.timeout(15_000) }),
  config: () => request<Config>("/api/config"),
  exchangeRates: () => request<ExchangeRates>("/api/exchange-rates"),
-  wakeServers: (serverIds: string[]) => request<void>("/api/live/wake", {
-    method: "POST",
-    body: JSON.stringify({ server_ids: serverIds }),
-    signal: AbortSignal.timeout(15_000),
-  }),
  refreshExchangeRates: () => request<ExchangeRates>("/api/admin/exchange-rates/refresh", { method: "POST" }, true),
  adminServers: () => request<{ servers: AdminServer[] }>("/api/admin/servers", {}, true),
  history: (id: string, hours: number) =>
@@ -115,7 +115,7 @@ const emptyServer: ServerInput = {
  network_interface: "",
  reset_day: 1,
  report_interval: 60,
-  collect_interval: 1,
+  collect_interval: 3,
  rx_correction: 0,
  tx_correction: 0,
  agent_mirror: "",
@@ -1309,7 +1309,7 @@ export function AdminPanel({
        <div className="form-grid"><label><span>名称</span><input autoFocus required value={form.name} onChange={(event) => updateForm("name", event.target.value)} /></label><label><span>地区代码</span><input maxLength={16} placeholder="CN / JP / DE" value={form.region} onChange={(event) => updateForm("region", event.target.value.toUpperCase())} /></label><label><span>分组</span><input value={form.group_name} onChange={(event) => updateForm("group_name", event.target.value)} /></label><label><span>标签</span><input placeholder="主力, 线路:BGP" value={form.tags} onChange={(event) => updateForm("tags", event.target.value)} /></label></div>
        <div className="form-grid three"><label><span>流量限额（0 不限）</span><input placeholder="如 100 G，不带单位按 GB；0 不限" {...sizeInputProps(trafficLimitText, setTrafficLimitText, (bytes) => updateForm("traffic_limit", bytes), form.traffic_limit)} /></label><label><span>流量口径</span><select value={form.traffic_limit_type} onChange={(event) => updateForm("traffic_limit_type", event.target.value as ServerInput["traffic_limit_type"])}><option value="sum">上下行合计</option><option value="max">取较大值</option><option value="min">取较小值</option><option value="up">仅上行</option><option value="down">仅下行</option></select></label><label><span>流量重置日</span><input min="1" max="31" type="number" value={form.reset_day} onChange={(event) => updateForm("reset_day", Number(event.target.value))} /></label></div>
        <div className="form-grid three"><label><span>价格（0 免费）</span><input type="number" min="0" max="1000000000" step="any" inputMode="decimal" value={priceText} onChange={(event) => updatePriceText(event.target.value)} onBlur={() => setPriceText(String(form.price))} /></label><label><span>币种</span><select value={form.currency} onChange={(event) => updateForm("currency", event.target.value)}>{ASSET_CURRENCIES.map((code) => <option key={code}>{code}</option>)}</select></label><label><span>计费周期</span><select value={String(form.billing_cycle)} onChange={(event) => updateForm("billing_cycle", Number(event.target.value))}>{BILLING_CYCLES.map((cycle) => <option key={cycle.days} value={cycle.days}>{cycle.label}</option>)}{BILLING_CYCLES.every((cycle) => cycle.days !== form.billing_cycle) ? <option value={form.billing_cycle}>{form.billing_cycle} 天</option> : null}</select></label></div>
-        <div className="form-grid three"><label><span>到期日期</span><input type="date" value={formatDate(form.expires_at)} onChange={(event) => updateForm("expires_at", event.target.value ? Math.floor(new Date(`${event.target.value}T00:00:00Z`).getTime() / 1000) : null)} /></label><label><span>历史保存间隔（秒）</span><input min="15" max="3600" type="number" value={form.report_interval} onChange={(event) => updateForm("report_interval", Number(event.target.value))} /></label><label><span>实时采样间隔（秒）</span><input min="1" max="60" type="number" value={form.collect_interval} onChange={(event) => updateForm("collect_interval", Number(event.target.value))} /></label></div>
+        <div className="form-grid three"><label><span>到期日期</span><input type="date" value={formatDate(form.expires_at)} onChange={(event) => updateForm("expires_at", event.target.value ? Math.floor(new Date(`${event.target.value}T00:00:00Z`).getTime() / 1000) : null)} /></label><label><span>历史保存间隔（秒）</span><input min="15" max="3600" type="number" value={form.report_interval} onChange={(event) => updateForm("report_interval", Number(event.target.value))} /></label><label><span>实时采样间隔（秒）</span><input min="3" max="60" type="number" value={form.collect_interval} onChange={(event) => updateForm("collect_interval", Number(event.target.value))} /></label></div>
        <div className="form-grid"><label><span>统计网卡（逗号分隔，留空自动）</span><input value={form.network_interface} onChange={(event) => updateForm("network_interface", event.target.value)} placeholder="eth0,ens3" /></label><label><span>Agent 下载加速（可选）</span><input value={form.agent_mirror} onChange={(event) => updateForm("agent_mirror", event.target.value.trim())} placeholder="https://ghproxy.net" /></label><label><span>上行流量当前值</span><input placeholder="如 500 G" {...sizeInputProps(txCurrentText, setTxCurrentText, setTxCurrentBytes, txCurrentBytes)} /></label><label><span>下行流量当前值</span><input placeholder="如 500 G" {...sizeInputProps(rxCurrentText, setRxCurrentText, setRxCurrentBytes, rxCurrentBytes)} /></label></div>
        <div className="settings-toggles editor-toggles"><Toggle label={form.billing_cycle <= 0 ? "自动续费（一次性不适用）" : "自动续费"} checked={form.auto_renewal} onChange={(value) => updateForm("auto_renewal", value)} /><Toggle label="Agent 自动更新" checked={form.auto_update} onChange={(value) => updateForm("auto_update", value)} /><Toggle label="隐藏节点" checked={form.hidden} onChange={(value) => updateForm("hidden", value)} /><Toggle label="关闭离线告警" checked={form.offline_notify_disabled} onChange={(value) => updateForm("offline_notify_disabled", value)} /></div>
        <div className="form-actions"><button type="button" className="secondary-btn" onClick={() => setEditing(null)}>取消</button><button className="primary-btn" disabled={busy}><Save size={15} />保存节点</button></div>
@@ -94,10 +94,11 @@ function CarrierPanel({ label, rows, kind, empty }: {
  );
 }

-export function NodeCard({ server, config, liveLatencyResults, onOpen }: {
+export function NodeCard({ server, config, liveLatencyResults, liveConnected, onOpen }: {
  server: Server;
  config: Config;
  liveLatencyResults?: LiveLatencyResult[];
+  liveConnected: boolean;
  onOpen: () => void;
 }) {
  const threshold = config.offline_threshold_seconds;
@@ -108,7 +109,7 @@ export function NodeCard({ server, config, liveLatencyResults, onOpen }: {
  const traffic = server.traffic_limit > 0 ? Math.min(100, (usedTraffic / server.traffic_limit) * 100) : 0;
  const locale = config.locale;
  const showCarriers = themeToggle(config, "showCarrierLatency", false);
-  const quality = useNodeLatency(server, config.show_latency, locale, showCarriers ? carrierSelection(config) : null, liveLatencyResults);
+  const quality = useNodeLatency(server, config.show_latency, locale, showCarriers ? carrierSelection(config) : null, liveLatencyResults, liveConnected);
  const price = formatPrice(server, locale);
  const remainingValue = remainingAssetValue(server.price, server.billing_cycle, server.expires_at);
  const showExpiryPanel = config.show_expiry || config.show_price;
@@ -39,10 +39,21 @@ export interface NodeLatencyStats {
 }

 const cache = new Map<string, { at: number; signature: string; points: LatencySample[] }>();
+const requests = new Map<string, Promise<{ points: LatencySample[] }>>();
 const CACHE_TTL = 120_000;
 const REFRESH_INTERVAL = 120_000;
 const WINDOW_HOURS = 1;

+function fetchHistory(id: string, signature: string) {
+  const key = `${id}:${signature}`;
+  let request = requests.get(key);
+  if (!request) {
+    request = api.latencyHistory(id, WINDOW_HOURS).finally(() => requests.delete(key));
+    requests.set(key, request);
+  }
+  return request;
+}
+
 function mergePoints(...sources: LatencySample[][]): LatencySample[] {
  const points = new Map<string, LatencySample>();
  for (const source of sources) {
@@ -69,11 +80,12 @@ export function useNodeLatency(
  locale: UiLocale,
  carrierSelection: CarrierSlotKey | null = null,
  liveResults?: LiveLatencyResult[],
+  liveConnected = false,
 ): NodeLatencyStats {
  const [fetched, setFetched] = useState<LatencySample[]>(server.latency);
  const [loading, setLoading] = useState(enabled);

-  const taskSignature = [...new Set(server.latency.map((point) => point.task_id))].sort().join(",");
+  const taskSignature = JSON.stringify(server.latency.map((point) => [point.task_id, point.name, point.task_type, point.target, point.port]));

  useEffect(() => {
    setFetched(server.latency);
@@ -83,6 +95,7 @@ export function useNodeLatency(
    if (!enabled) { setLoading(false); return; }
    let stopped = false;
    let running = false;
+    let loaded = false;
    let timer: number | undefined;

    const canRefresh = () => !document.hidden && navigator.onLine !== false;
@@ -94,7 +107,7 @@ export function useNodeLatency(
    };
    const schedule = () => {
      clearTimer();
-      if (!stopped && canRefresh()) {
+      if (!stopped && canRefresh() && (!liveConnected || !loaded)) {
        timer = window.setTimeout(() => void load(true), REFRESH_INTERVAL);
      }
    };
@@ -107,10 +120,12 @@ export function useNodeLatency(
        if (!force && hit?.signature === taskSignature && Date.now() - hit.at < CACHE_TTL) {
          setFetched(hit.points);
          setLoading(false);
+          loaded = true;
          return;
        }
        if (!hit) setLoading(true);
-        const result = await api.latencyHistory(server.id, WINDOW_HOURS);
+        const result = await fetchHistory(server.id, taskSignature);
+        loaded = true;
        if (!stopped) {
          cache.set(server.id, { at: Date.now(), signature: taskSignature, points: result.points });
          setFetched(result.points);
@@ -134,7 +149,7 @@ export function useNodeLatency(
    document.addEventListener("visibilitychange", handleVisibility);
    window.addEventListener("online", resume);
    window.addEventListener("offline", pause);
-    if (canRefresh()) void load(false);
+    if (canRefresh()) void load(liveConnected);
    return () => {
      stopped = true;
      clearTimer();
@@ -142,7 +157,7 @@ export function useNodeLatency(
      window.removeEventListener("online", resume);
      window.removeEventListener("offline", pause);
    };
-  }, [enabled, server.id, taskSignature]);
+  }, [enabled, liveConnected, server.id, taskSignature]);

  const points = useMemo(() => {
    const cutoff = Date.now() / 1000 - WINDOW_HOURS * 3600;
@@ -1,7 +1,6 @@
-import { describe, expect, spyOn, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import {
  BOOTSTRAP_LIVE_SYNC_INTERVAL_MS,
-  createLiveFlushScheduler,
  createRefreshQueue,
  hasActiveRemoteTasks,
  isRemoteTaskActive,
@@ -39,68 +38,6 @@ describe("createRefreshQueue", () => {
  });
 });

-describe("createLiveFlushScheduler", () => {
-  test("uses a fixed one-second clock, coalesces bursts and skips idle ticks", () => {
-    let flushes = 0;
-    let tick = () => {};
-    const timer = 123 as unknown as ReturnType<typeof setInterval>;
-    const start = spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, delay?: number) => {
-      expect(delay).toBe(1_000);
-      tick = callback as () => void;
-      return timer;
-    }) as typeof setInterval);
-    const stop = spyOn(globalThis, "clearInterval").mockImplementation(() => {});
-    try {
-      const scheduler = createLiveFlushScheduler(() => { flushes += 1; });
-      scheduler.schedule();
-      scheduler.schedule();
-      scheduler.schedule();
-      expect(flushes).toBe(0);
-      expect(start).toHaveBeenCalledTimes(1);
-      tick();
-      expect(flushes).toBe(1);
-      tick();
-      expect(flushes).toBe(1);
-      scheduler.schedule();
-      expect(start).toHaveBeenCalledTimes(1);
-      tick();
-      expect(flushes).toBe(2);
-      scheduler.cancel();
-      expect(stop).toHaveBeenCalledWith(timer);
-    } finally {
-      start.mockRestore();
-      stop.mockRestore();
-    }
-  });
-
-  test("cancel drops pending data and allows restarting the clock", () => {
-    let flushes = 0;
-    let tick = () => {};
-    const start = spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void) => {
-      tick = callback as () => void;
-      return 123 as unknown as ReturnType<typeof setInterval>;
-    }) as typeof setInterval);
-    const stop = spyOn(globalThis, "clearInterval").mockImplementation(() => {});
-    try {
-      const scheduler = createLiveFlushScheduler(() => { flushes += 1; });
-      scheduler.schedule();
-      scheduler.cancel();
-      scheduler.cancel();
-      tick();
-      expect(flushes).toBe(0);
-      expect(stop).toHaveBeenCalledTimes(1);
-      scheduler.schedule();
-      expect(start).toHaveBeenCalledTimes(2);
-      tick();
-      expect(flushes).toBe(1);
-      scheduler.cancel();
-    } finally {
-      start.mockRestore();
-      stop.mockRestore();
-    }
-  });
-});
-
 describe("bootstrap refresh policy", () => {
  test("refreshes every tick while the live connection is down", () => {
    expect(shouldSyncBootstrap(false, 0)).toBe(true);
@@ -4,34 +4,6 @@ export const BOOTSTRAP_POLL_INTERVAL_MS = 15_000;
 export const BOOTSTRAP_LIVE_SYNC_INTERVAL_MS = 5 * 60_000;
 export const REMOTE_TASK_POLL_INTERVAL_MS = 2_000;
 export const REMOTE_TASK_POLL_TIMEOUT_MS = 60_000;
-const LIVE_FLUSH_INTERVAL_MS = 1_000;
-
-export interface LiveFlushScheduler {
-  schedule: () => void;
-  cancel: () => void;
-}
-
-export function createLiveFlushScheduler(flush: () => void, intervalMs = LIVE_FLUSH_INTERVAL_MS): LiveFlushScheduler {
-  let timer: ReturnType<typeof setInterval> | undefined;
-  let pending = false;
-  return {
-    schedule: () => {
-      pending = true;
-      if (timer !== undefined) return;
-      timer = setInterval(() => {
-        if (!pending) return;
-        pending = false;
-        flush();
-      }, intervalMs);
-    },
-    cancel: () => {
-      pending = false;
-      if (timer === undefined) return;
-      clearInterval(timer);
-      timer = undefined;
-    },
-  };
-}

 export function createRefreshQueue(execute: (quiet: boolean) => Promise<void>) {
  let active: Promise<void> | null = null;
@@ -1,6 +1,13 @@
 import { describe, expect, spyOn, test } from "bun:test";
 import { connectLive, reconnectDelay } from "./transport";
 import type { BatchUpdate } from "./live";
+import { gzipSync } from "node:zlib";
+
+function compressed(value: unknown): ArrayBuffer {
+  return Uint8Array.from(gzipSync(JSON.stringify(value))).buffer;
+}
+
+async function settle() { await new Promise((resolve) => setTimeout(resolve, 20)); }

 describe("reconnectDelay", () => {
  test("backs off exponentially", () => {
@@ -31,14 +38,15 @@ class FakeSocket {
  onopen: (() => void) | null = null;
  onclose: (() => void) | null = null;
  onerror: (() => void) | null = null;
-  onmessage: ((event: { data: string }) => void) | null = null;
+  binaryType = "blob";
+  onmessage: ((event: { data: string | ArrayBuffer }) => void) | null = null;
  sent: string[] = [];
  failSend = false;

  constructor(public url: URL) { FakeSocket.instances.push(this); }
  open() { this.readyState = FakeSocket.OPEN; this.onopen?.(); }
  close() { this.readyState = FakeSocket.CLOSED; this.onclose?.(); }
-  receive(data: string) { this.onmessage?.({ data }); }
+  receive(data: string | ArrayBuffer) { this.onmessage?.({ data }); }
  send(data: string) {
    if (this.failSend) throw new Error("Disconnected");
    this.sent.push(data);
@@ -70,7 +78,6 @@ function transportHarness() {
  const dispose = connectLive({ serverId: "node/1" }, {
    onBatch: (batch) => { batches.push(batch); },
    onConnectedChange: (connected) => { connections.push(connected); },
-    onWakeRequested: async () => {},
  });
  return {
    sockets: FakeSocket.instances,
@@ -172,7 +179,7 @@ describe("live connection lifecycle", () => {
    } finally { h.restore(); }
  });

-  test("ignores callbacks from closed sockets and accepts new batches", () => {
+  test("ignores callbacks from closed sockets and decodes batches in wire order", async () => {
    const h = transportHarness();
    try {
      const staleMessage = h.sockets[0].onmessage!;
@@ -184,8 +191,40 @@ describe("live connection lifecycle", () => {
      expect(h.connections).toEqual([]);
      expect(h.batches).toEqual([]);
      h.sockets[1].open();
-      h.sockets[1].receive('{"type":"batchUpdate","updates":[{"serverId":"new"}]}');
-      expect(h.batches).toEqual([[{ serverId: "new" }]]);
+      expect(h.sockets[1].binaryType).toBe("arraybuffer");
+      h.sockets[1].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "first" }] }));
+      h.sockets[1].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "second" }] }));
+      await settle();
+      expect(h.batches).toEqual([[{ serverId: "first" }], [{ serverId: "second" }]]);
+    } finally { h.restore(); }
+  });
+
+  test("drops queued decoding after disconnect and reconnects on invalid gzip", async () => {
+    const h = transportHarness();
+    try {
+      h.sockets[0].open();
+      h.sockets[0].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "stale" }] }));
+      h.visibility(true);
+      await settle();
+      expect(h.batches).toEqual([]);
+      h.visibility(false);
+      h.sockets[1].open();
+      h.sockets[1].receive(new Uint8Array([1, 2, 3]).buffer);
+      await settle();
+      expect(h.sockets[1].readyState).toBe(FakeSocket.CLOSED);
+      h.advance(1_000);
+      expect(h.sockets).toHaveLength(3);
+    } finally { h.restore(); }
+  });
+
+  test("rejects decompression bombs", async () => {
+    const h = transportHarness();
+    try {
+      h.sockets[0].open();
+      h.sockets[0].receive(compressed("x".repeat(2 * 1024 * 1024 + 1)));
+      await settle();
+      expect(h.sockets[0].readyState).toBe(FakeSocket.CLOSED);
+      expect(h.batches).toEqual([]);
    } finally { h.restore(); }
  });
 });
@@ -7,6 +7,30 @@ const STABLE_CONNECTION_MS = 10_000;
 const CONNECTION_TIMEOUT = 10_000;
 const HEARTBEAT_INTERVAL = 30_000;
 const HEARTBEAT_TIMEOUT = 10_000;
+const MAX_MESSAGE_BYTES = 2 * 1024 * 1024;
+
+async function decodeMessage(data: ArrayBuffer) {
+  if (data.byteLength > MAX_MESSAGE_BYTES) throw new Error("Live message too large");
+  const reader = new Blob([data]).stream().pipeThrough(new DecompressionStream("gzip")).getReader();
+  const chunks: Uint8Array[] = [];
+  let size = 0;
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      size += value.byteLength;
+      if (size > MAX_MESSAGE_BYTES) throw new Error("Expanded live message too large");
+      chunks.push(value);
+    }
+    const bytes = new Uint8Array(size);
+    let offset = 0;
+    for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; }
+    return JSON.parse(new TextDecoder().decode(bytes));
+  } finally {
+    await reader.cancel().catch(() => {});
+    reader.releaseLock();
+  }
+}

 export function reconnectDelay(attempt: number, randomValue = Math.random()) {
  const exponent = Math.max(0, Math.min(30, Math.floor(attempt)));
@@ -19,7 +43,6 @@ export function reconnectDelay(attempt: number, randomValue = Math.random()) {
 export interface LiveTransportHandlers {
  onBatch: (updates: BatchUpdate[]) => void;
  onConnectedChange: (connected: boolean) => void;
-  onWakeRequested: () => Promise<void>;
 }

 export interface LiveTransportOptions {
@@ -48,7 +71,6 @@ export function connectLive(
  let reconnectAttempt = 0;
  let connected = false;
  let openedAt: number | null = null;
-  let wakeInFlight = false;

  const setConnected = (next: boolean) => {
    if (connected === next) return;
@@ -62,15 +84,6 @@ export function connectLive(
    reconnectTimer = null;
  };

-  const wakeAgents = () => {
-    if (cancelled || suspended || wakeInFlight) return;
-    wakeInFlight = true;
-    void Promise.resolve()
-      .then(() => { if (!cancelled && !suspended) return handlers.onWakeRequested(); })
-      .catch(() => {})
-      .finally(() => { wakeInFlight = false; });
-  };
-
  const scheduleReconnect = () => {
    if (cancelled || suspended || reconnectTimer !== null) return;
    const delay = reconnectDelay(reconnectAttempt);
@@ -111,7 +124,6 @@ export function connectLive(
    deadlineTimer = window.setTimeout(() => closeCurrentSocket(true), HEARTBEAT_TIMEOUT);
    try {
      socket.send("ping");
-      wakeAgents();
    } catch {
      closeCurrentSocket(true);
    }
@@ -128,6 +140,9 @@ export function connectLive(
      return;
    }
    socket = current;
+    current.binaryType = "arraybuffer";
+    let messages = Promise.resolve();
+    let queuedBytes = 0;
    deadlineTimer = window.setTimeout(() => closeCurrentSocket(true), CONNECTION_TIMEOUT);
    current.onopen = () => {
      if (socket !== current || cancelled || suspended) return;
@@ -152,12 +167,20 @@ export function connectLive(
        heartbeatTimer = window.setTimeout(heartbeat, HEARTBEAT_INTERVAL);
        return;
      }
-      try {
-        const message = JSON.parse(event.data);
-        if (message.type === "batchUpdate" && Array.isArray(message.updates)) {
+      if (!(event.data instanceof ArrayBuffer)) return;
+      const data = event.data;
+      queuedBytes += data.byteLength;
+      if (queuedBytes > MAX_MESSAGE_BYTES * 4) { closeCurrentSocket(true); return; }
+      // Decompression is asynchronous; preserve wire order and drop stale results after reconnects.
+      messages = messages.then(async () => {
+        if (socket !== current) return;
+        const message = await decodeMessage(data);
+        if (socket === current && message.type === "batchUpdate" && Array.isArray(message.updates)) {
          handlers.onBatch(message.updates as BatchUpdate[]);
        }
-      } catch {}
+      }).catch(() => {
+        if (socket === current) closeCurrentSocket(true);
+      }).finally(() => { queuedBytes -= data.byteLength; });
    };
  };

@@ -9,6 +9,7 @@ import { tmpdir } from "node:os";
 import { join, resolve } from "node:path";
 import { setTimeout as delay } from "node:timers/promises";
 import { WebSocketServer } from "ws";
+import { gunzipSync } from "node:zlib";

 const binary = resolve(process.env.MONITOR_AGENT_BINARY ?? "agent/target/debug/nodeflare-agent");
 const directory = mkdtempSync(join(tmpdir(), "nodeflare-agent-runtime-"));
@@ -67,7 +68,7 @@ try {
  proxy.listen(0, "127.0.0.1");
  await once(proxy, "listening");

-  console.log("agent runtime: automatic update during one-second sampling");
+  console.log("agent runtime: automatic update during three-second sampling");
  websocketServer = new WebSocketServer({ noServer: true });
  upgradeServer = createHttpServer();
  let redirected = false;
@@ -89,27 +90,30 @@ try {
    socket.send(JSON.stringify({
      type: "config", ts: Math.floor(Date.now() / 1_000),
      config: {
-        report_interval: 15, collect_interval: 1, network_interface: "",
+        report_interval: 15, collect_interval: 3, network_interface: "",
        agent_mirror: "", auto_update: true, latency_tasks: [],
      },
    }));
-    socket.on("message", (raw) => {
-      const message = JSON.parse(raw.toString());
-      if (message.type !== "update") return;
+    let receivedThrough = 0;
+    socket.on("message", (raw, binary) => {
+      if (!binary) return;
+      const message = JSON.parse(gunzipSync(raw).toString());
      samples += message.samples.length;
-      if (message.persist) persistedThrough = Math.max(...message.samples.map((sample) => sample.timestamp));
+      receivedThrough = Math.max(receivedThrough, ...message.samples.map((sample) => sample.metrics.timestamp));
+      if (!message.persist) return;
+      persistedThrough = receivedThrough;
      socket.send(JSON.stringify({
        type: "ack", ts: Math.floor(Date.now() / 1_000),
        persisted: message.persist, persistenceError: false,
        persistedThroughTs: persistedThrough,
-        nextPersistAfterMs: 1_000, nextWssReportAfterMs: 1_000, realtimeHint: false,
+        nextPersistAfterMs: 15_000,
      }));
    });
  });
  let token;
  for (let index = 0; index < 100_000; index++) {
    const candidate = `runtime-update-${index}`;
-    if (createHash("sha256").update(candidate).digest().readBigUInt64BE(0) % 1_801n === 2n) {
+    if (createHash("sha256").update(candidate).digest().readBigUInt64BE(0) % 1_801n === 5n) {
      token = candidate;
      break;
    }
@@ -121,6 +125,40 @@ try {
  assert(redirected, "Same-origin WebSocket redirects must remain supported");
  await stopAgent(updating);

+  console.log("agent runtime: three-second uploads are unique and reconnect replays unconfirmed samples");
+  const telemetryConnections = [];
+  const trackTelemetry = (socket) => {
+    const connection = { timestamps: [], durable: 0, firstInfo: false, infoCount: 0, commits: 0 };
+    telemetryConnections.push(connection);
+    socket.on("message", (raw, binary) => {
+      if (!binary) return;
+      const message = JSON.parse(gunzipSync(raw).toString());
+      for (const sample of message.samples) {
+        if (!connection.timestamps.length) connection.firstInfo = Boolean(sample.info);
+        connection.infoCount += Number(Boolean(sample.info));
+        assert(!connection.timestamps.includes(sample.metrics.timestamp), "Sample was uploaded twice on the same connection");
+        connection.timestamps.push(sample.metrics.timestamp);
+      }
+      if (message.persist) {
+        connection.durable = connection.timestamps.at(-1);
+        connection.commits++;
+      }
+      if (telemetryConnections.length === 1 && connection.timestamps.length === 3) socket.terminate();
+    });
+  };
+  websocketServer.on("connection", trackTelemetry);
+  const telemetry = startAgent(upgradeServer.address().port, "runtime-telemetry", "telemetry");
+  await waitUntil(() => telemetryConnections.length >= 2 && telemetryConnections[1].commits >= 2,
+    40_000, "Telemetry reconnect or commit did not complete");
+  await stopAgent(telemetry);
+  websocketServer.off("connection", trackTelemetry);
+  const [before, after] = telemetryConnections;
+  assert(before.firstInfo && after.firstInfo, "Static info must be resent on reconnect");
+  assert.equal(before.infoCount, 1, "Unchanged static info must not be uploaded repeatedly");
+  assert(before.timestamps.slice(1).every((timestamp) => after.timestamps.includes(timestamp)), "Unconfirmed samples were lost");
+  assert(!after.timestamps.includes(before.durable), "Durable samples must not be replayed");
+  assert(before.timestamps.slice(1).every((timestamp, index) => timestamp - before.timestamps[index] >= 3), "Sampling interval is shorter than three seconds");
+
  console.log("agent runtime: remote command survives disconnect and returns its result");
  const taskId = randomUUID();
  let remoteConnections = 0;
@@ -128,7 +166,8 @@ try {
  let remoteResult;
  websocketServer.on("connection", (socket) => {
    remoteConnections++;
-    socket.on("message", (raw) => {
+    socket.on("message", (raw, binary) => {
+      if (binary) return;
      const message = JSON.parse(raw.toString());
      if (message.type === "task_received" && message.task_id === taskId) {
        received = true;
@@ -272,7 +272,7 @@ MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \
 if [ -n "${MONITOR_AGENT_BINARY:-}" ]; then
  step "real Agent failed-latency and persistence"
  request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' -X PATCH \
-    --data "$(printf '%s' "$server_input" | jq '.report_interval=15 | .collect_interval=1')" \
+    --data "$(printf '%s' "$server_input" | jq '.report_interval=15 | .collect_interval=3')" \
    "$MONITOR_BASE_URL/api/admin/servers/$server_id" >/dev/null
  agent_latency_task_id=$(request -H "Authorization: Bearer $admin_token" \
    -H 'Content-Type: application/json' \
@@ -1,6 +1,22 @@
 import WebSocket from "ws";
 import assert from "node:assert/strict";
 import { once } from "node:events";
+import { gzipSync, gunzipSync } from "node:zlib";
+
+function telemetryPayload(reports, persist) {
+  return gzipSync(JSON.stringify({
+    samples: reports.map((report) => {
+      const metrics = { ...report };
+      const info = {};
+      for (const key of ["cpu_model", "os", "kernel", "arch", "virtualization", "gpu_model", "agent_version", "ip_v4", "ip_v6"]) {
+        info[key] = metrics[key] ?? "";
+        delete metrics[key];
+      }
+      return { metrics, info };
+    }),
+    persist,
+  }));
+}

 const baseUrl = process.env.MONITOR_BASE_URL;
 const adminToken = process.env.MONITOR_ADMIN_TOKEN;
@@ -190,10 +206,10 @@ function waitForJsonMessage(socket, expected, predicate) {
      cleanup();
      reject(error);
    };
-    const onMessage = (data) => {
+    const onMessage = (data, binary) => {
      let message;
      try {
-        message = JSON.parse(data.toString());
+        message = JSON.parse((binary ? gunzipSync(data) : data).toString());
      } catch {
        return;
      }
@@ -358,7 +374,7 @@ try {
      name: "Offline remote smoke", region: "", group_name: "", tags: "", hidden: true,
      expires_at: null, traffic_limit: 0, traffic_limit_type: "max", price: 0,
      billing_cycle: 30, currency: "USD", auto_renewal: false, network_interface: "",
-      reset_day: 1, report_interval: 60, collect_interval: 1, rx_correction: 0,
+      reset_day: 1, report_interval: 60, collect_interval: 3, rx_correction: 0,
      tx_correction: 0, agent_mirror: "", offline_notify_disabled: true, auto_update: false,
    }),
  });
@@ -554,27 +570,6 @@ try {
    });
  }

-  const wakeHintPromise = waitForJsonMessage(
-    agent,
-    "batched overview wake hint",
-    (message) => message.type === "ack" && message.realtimeHint === true,
-  );
-  const wakeResponse = await fetch(new URL("/api/live/wake", baseUrl), {
-    method: "POST",
-    headers: {
-      Authorization: `Bearer ${adminToken}`,
-      "Content-Type": "application/json",
-    },
-    body: JSON.stringify({ server_ids: [serverId] }),
-  });
-  if (wakeResponse.status !== 204) {
-    throw new Error(`Batch wake returned HTTP ${wakeResponse.status}: ${await wakeResponse.text()}`);
-  }
-  const wakeHint = await wakeHintPromise;
-  if (wakeHint.nextWssReportAfterMs !== 5_000) {
-    throw new Error(`Invalid batch wake hint: ${JSON.stringify(wakeHint)}`);
-  }
-
  const timestamp = Math.floor(Date.now() / 1_000);
  const baseSample = {
    timestamp,
@@ -654,22 +649,10 @@ try {
    sample(5, { net_rx_total: 536870912, net_tx_total: 268435456 }),
  ];
  const latestTimestamp = timestamp + 5;
-  for (const persist of [undefined, null]) {
-    const rejectedAck = waitForJsonMessage(
-      agent,
-      "invalid persistence flag ACK",
-      (message) => message.type === "ack" && message.realtimeHint === false,
-    );
-    agent.send(JSON.stringify({ type: "update", batchId: "invalid-persist", samples, persist }));
-    const rejected = await rejectedAck;
-    if (rejected.persistenceError !== true || rejected.persisted !== false) {
-      throw new Error(`Accepted invalid persistence flag: ${JSON.stringify(rejected)}`);
-    }
-  }
  const ackPromise = waitForJsonMessage(
    agent,
    "Agent metric ACK",
-    (message) => message.type === "ack" && message.realtimeHint === false,
+    (message) => message.type === "ack",
  );
  const updatePromise = waitForJsonMessage(
    dashboard,
@@ -684,7 +667,7 @@ try {
          ),
      ),
  );
-  agent.send(JSON.stringify({ type: "update", batchId: `smoke-${timestamp}`, samples, persist: true }));
+  agent.send(telemetryPayload(samples, true));
  const [ack, update] = await Promise.all([ackPromise, updatePromise]);
  if (
    ack.ts <= 0 ||
@@ -692,11 +675,12 @@ try {
    ack.persistenceError !== false ||
    !Number.isInteger(ack.persistedThroughTs) ||
    ack.persistedThroughTs < latestTimestamp ||
-    ack.nextPersistAfterMs !== 60_000 ||
-    ack.nextWssReportAfterMs !== 5_000
+    ack.nextPersistAfterMs !== 60_000
  ) {
    throw new Error(`Invalid Agent metric ACK: ${JSON.stringify(ack)}`);
  }
+  assert.equal(update.updates[0].samples.length, 1, "Browser receives only the latest metric snapshot");
+  assert.equal(update.updates[0].samples[0].data.latency_results.length, 2, "All new latency results are retained");

  const publicIdentityFields = [
    "cpu_model",
@@ -0,0 +1,162 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "adler2"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
+
+[[package]]
+name = "cfg-if"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
+
+[[package]]
+name = "crc32fast"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8498c871161e1742aaa9d52551b2d6ebdd4c3d45a3be423e3728f33b955be550"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "flate2"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
+dependencies = [
+ "crc32fast",
+ "miniz_oxide",
+ "zlib-rs",
+]
+
+[[package]]
+name = "itoa"
+version = "1.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+
+[[package]]
+name = "memchr"
+version = "2.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
+
+[[package]]
+name = "miniz_oxide"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
+dependencies = [
+ "adler2",
+ "simd-adler32",
+]
+
+[[package]]
+name = "nodeflare-telemetry"
+version = "1.0.0"
+dependencies = [
+ "flate2",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.107"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.47"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "serde"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
+dependencies = [
+ "serde_core",
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_core"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.151"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
+dependencies = [
+ "itoa",
+ "memchr",
+ "serde",
+ "serde_core",
+ "zmij",
+]
+
+[[package]]
+name = "simd-adler32"
+version = "0.3.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea"
+
+[[package]]
+name = "syn"
+version = "3.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+
+[[package]]
+name = "zlib-rs"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12"
+
+[[package]]
+name = "zmij"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
@@ -0,0 +1,14 @@
+[package]
+name = "nodeflare-telemetry"
+version = "1.0.0"
+edition = "2024"
+license = "MIT"
+publish = false
+
+[lib]
+path = "telemetry.rs"
+
+[dependencies]
+flate2 = "1"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
@@ -0,0 +1,347 @@
+use flate2::{Compression, read::GzDecoder, write::GzEncoder};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
+use std::io::{self, Read};
+
+pub const MIN_COLLECT_INTERVAL: u64 = 3;
+pub const MAX_BATCH_BYTES: usize = 768 * 1024;
+pub const MAX_DECODED_BYTES: usize = 2 * 1024 * 1024;
+pub const MAX_SAMPLES: usize = 720;
+
+#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct Report {
+    pub timestamp: i64,
+    pub cpu: f64,
+    pub load1: f64,
+    pub load5: f64,
+    pub load15: f64,
+    pub mem_used: i64,
+    pub mem_total: i64,
+    pub swap_used: i64,
+    pub swap_total: i64,
+    pub disk_used: i64,
+    pub disk_total: i64,
+    pub net_in: f64,
+    pub net_out: f64,
+    pub net_rx_total: i64,
+    pub net_tx_total: i64,
+    pub uptime: i64,
+    pub processes: i64,
+    pub tcp_connections: i64,
+    pub udp_connections: i64,
+    pub cpu_cores: i64,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub cpu_model: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub os: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub kernel: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub arch: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub virtualization: String,
+    pub gpu_usage: f64,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub gpu_model: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub agent_version: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub ip_v4: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub ip_v6: String,
+    pub disk_read_bps: f64,
+    pub disk_write_bps: f64,
+    pub disk_read_iops: f64,
+    pub disk_write_iops: f64,
+    pub disk_await_ms: f64,
+    pub disk_utilization: f64,
+    pub disks: Vec<DiskMetric>,
+    pub gpus: Vec<GpuMetric>,
+    pub latency_results: Vec<LatencyResult>,
+}
+
+impl Report {
+    fn round_for_upload(&mut self) {
+        for value in [
+            &mut self.net_in,
+            &mut self.net_out,
+            &mut self.disk_read_bps,
+            &mut self.disk_write_bps,
+        ] {
+            *value = value.round();
+        }
+        for value in [
+            &mut self.cpu,
+            &mut self.load1,
+            &mut self.load5,
+            &mut self.load15,
+            &mut self.gpu_usage,
+            &mut self.disk_read_iops,
+            &mut self.disk_write_iops,
+            &mut self.disk_await_ms,
+            &mut self.disk_utilization,
+        ] {
+            *value = (*value * 100.0).round() / 100.0;
+        }
+        for disk in &mut self.disks {
+            disk.read_bps = disk.read_bps.round();
+            disk.write_bps = disk.write_bps.round();
+            for value in [
+                &mut disk.read_iops,
+                &mut disk.write_iops,
+                &mut disk.await_ms,
+                &mut disk.utilization,
+            ] {
+                *value = (*value * 100.0).round() / 100.0;
+            }
+        }
+        for gpu in &mut self.gpus {
+            gpu.usage = gpu.usage.map(|value| (value * 100.0).round() / 100.0);
+        }
+        for latency in &mut self.latency_results {
+            latency.latency_ms = (latency.latency_ms * 100.0).round() / 100.0;
+            latency.packet_loss = (latency.packet_loss * 100.0).round() / 100.0;
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct DiskMetric {
+    pub name: String,
+    pub mount_point: String,
+    pub used: i64,
+    pub total: i64,
+    pub read_bps: f64,
+    pub write_bps: f64,
+    pub read_iops: f64,
+    pub write_iops: f64,
+    pub await_ms: f64,
+    pub utilization: f64,
+}
+
+#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct GpuMetric {
+    pub model: String,
+    pub usage: Option<f64>,
+    pub memory_used: i64,
+    pub memory_total: i64,
+}
+
+#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct LatencyResult {
+    pub task_id: String,
+    pub timestamp: i64,
+    pub latency_ms: f64,
+    pub packet_loss: f64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Info {
+    cpu_model: String,
+    os: String,
+    kernel: String,
+    arch: String,
+    virtualization: String,
+    gpu_model: String,
+    agent_version: String,
+    ip_v4: String,
+    ip_v6: String,
+}
+
+impl Info {
+    fn take(report: &mut Report) -> Self {
+        Self {
+            cpu_model: std::mem::take(&mut report.cpu_model),
+            os: std::mem::take(&mut report.os),
+            kernel: std::mem::take(&mut report.kernel),
+            arch: std::mem::take(&mut report.arch),
+            virtualization: std::mem::take(&mut report.virtualization),
+            gpu_model: std::mem::take(&mut report.gpu_model),
+            agent_version: std::mem::take(&mut report.agent_version),
+            ip_v4: std::mem::take(&mut report.ip_v4),
+            ip_v6: std::mem::take(&mut report.ip_v6),
+        }
+    }
+
+    fn apply(&self, report: &mut Report) {
+        report.cpu_model.clone_from(&self.cpu_model);
+        report.os.clone_from(&self.os);
+        report.kernel.clone_from(&self.kernel);
+        report.arch.clone_from(&self.arch);
+        report.virtualization.clone_from(&self.virtualization);
+        report.gpu_model.clone_from(&self.gpu_model);
+        report.agent_version.clone_from(&self.agent_version);
+        report.ip_v4.clone_from(&self.ip_v4);
+        report.ip_v6.clone_from(&self.ip_v6);
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Sample {
+    pub metrics: Report,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub info: Option<Info>,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Update {
+    pub samples: Vec<Sample>,
+    pub persist: bool,
+}
+
+impl Update {
+    pub fn from_reports(reports: Vec<Report>, persist: bool, info: &mut Option<Info>) -> Self {
+        let samples = reports
+            .into_iter()
+            .map(|mut metrics| {
+                let current = Info::take(&mut metrics);
+                metrics.round_for_upload();
+                let changed = info.as_ref() != Some(&current);
+                if changed {
+                    *info = Some(current.clone());
+                }
+                Sample {
+                    metrics,
+                    info: changed.then_some(current),
+                }
+            })
+            .collect();
+        Self { samples, persist }
+    }
+
+    pub fn into_reports(self, info: &mut Option<Info>) -> io::Result<Vec<Report>> {
+        if self.samples.len() > MAX_SAMPLES {
+            return Err(io::Error::other("too many telemetry samples"));
+        }
+        self.samples
+            .into_iter()
+            .map(|mut sample| {
+                if let Some(current) = sample.info {
+                    *info = Some(current);
+                }
+                info.as_ref()
+                    .ok_or_else(|| io::Error::other("missing telemetry info"))?
+                    .apply(&mut sample.metrics);
+                Ok(sample.metrics)
+            })
+            .collect()
+    }
+}
+
+pub fn encode(value: &impl Serialize) -> io::Result<Vec<u8>> {
+    let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
+    serde_json::to_writer(&mut encoder, value)?;
+    encoder.finish()
+}
+
+pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> io::Result<T> {
+    if bytes.len() > MAX_DECODED_BYTES {
+        return Err(io::Error::other("telemetry frame is too large"));
+    }
+    let mut decoded = Vec::new();
+    GzDecoder::new(bytes)
+        .take(MAX_DECODED_BYTES as u64 + 1)
+        .read_to_end(&mut decoded)?;
+    if decoded.len() > MAX_DECODED_BYTES {
+        return Err(io::Error::other("expanded telemetry frame is too large"));
+    }
+    serde_json::from_slice(&decoded).map_err(io::Error::other)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn trims_float_noise_without_changing_byte_counters() {
+        let report = Report {
+            cpu: 100.0 / 3.0,
+            net_in: 12345.6789,
+            net_rx_total: i64::MAX,
+            disks: vec![DiskMetric {
+                read_bps: 125.6789,
+                await_ms: 1.234567,
+                ..DiskMetric::default()
+            }],
+            latency_results: vec![LatencyResult {
+                latency_ms: 28.4444,
+                packet_loss: 100.0 / 3.0,
+                ..LatencyResult::default()
+            }],
+            ..Report::default()
+        };
+        let encoded = encode(&Update::from_reports(vec![report], false, &mut None)).unwrap();
+        let reports = decode::<Update>(&encoded)
+            .unwrap()
+            .into_reports(&mut None)
+            .unwrap();
+        assert_eq!(reports[0].cpu, 33.33);
+        assert_eq!(reports[0].net_in, 12346.0);
+        assert_eq!(reports[0].net_rx_total, i64::MAX);
+        assert_eq!(reports[0].disks[0].read_bps, 126.0);
+        assert_eq!(reports[0].disks[0].await_ms, 1.23);
+        assert_eq!(reports[0].latency_results[0].latency_ms, 28.44);
+    }
+
+    #[test]
+    fn info_is_sent_on_connect_and_changes_including_during_replay() {
+        let mut encoder = None;
+        let mut decoder = None;
+        for (name, changed) in [
+            ("first", true),
+            ("first", false),
+            ("second", true),
+            ("first", true),
+        ] {
+            let update = Update::from_reports(
+                vec![Report {
+                    cpu_model: name.into(),
+                    cpu: 23.5,
+                    ..Report::default()
+                }],
+                false,
+                &mut encoder,
+            );
+            assert_eq!(update.samples[0].info.is_some(), changed);
+            assert!(update.samples[0].metrics.cpu_model.is_empty());
+            let bytes = encode(&update).unwrap();
+            let decoded: Update = decode(&bytes).unwrap();
+            let reports = decoded.into_reports(&mut decoder).unwrap();
+            assert_eq!(reports[0].cpu_model, name);
+            assert_eq!(reports[0].cpu, 23.5);
+        }
+        let update = Update::from_reports(vec![Report::default()], true, &mut None);
+        assert!(update.samples[0].info.is_some());
+    }
+
+    #[test]
+    fn rejects_missing_info_corruption_and_decompression_bombs() {
+        let update = Update {
+            samples: vec![Sample {
+                metrics: Report::default(),
+                info: None,
+            }],
+            persist: false,
+        };
+        assert!(update.into_reports(&mut None).is_err());
+        assert!(decode::<Update>(b"not gzip").is_err());
+        let bytes = encode(&"x".repeat(MAX_DECODED_BYTES + 1)).unwrap();
+        assert!(decode::<String>(&bytes).is_err());
+    }
+
+    #[test]
+    fn compresses_batches_and_supports_empty_commit_messages() {
+        let update = Update::from_reports(vec![Report::default(); 20], false, &mut None);
+        assert!(encode(&update).unwrap().len() < serde_json::to_vec(&update).unwrap().len() / 4);
+        let update = Update::from_reports(Vec::new(), true, &mut None);
+        let decoded: Update = decode(&encode(&update).unwrap()).unwrap();
+        assert!(decoded.persist);
+        assert!(decoded.samples.is_empty());
+    }
+}