diff --git a/cpctl/cmd/stats.go b/cpctl/cmd/stats.go
index c4846259..aa27eb12 100644
--- a/cpctl/cmd/stats.go
+++ b/cpctl/cmd/stats.go
@@ -26,7 +26,7 @@ func init() {
 	f.IntVarP(&statsCfg.count, "count", "n", 0,
 		"number of samples (0 = run forever)")
 	f.DurationVarP(&statsCfg.interval, "interval", "i", 2*time.Second,
-		"interval between samples")
+		"poll interval; cpworker refreshes stats every 5s, so samples are at least 5s apart")
 }
 
 var statsCmd = &cobra.Command{
@@ -109,7 +109,7 @@ func emitStatsRaw(out io.Writer, s cpworker.StatsSummary, format string) error {
 	switch format {
 	case FormatJSONL:
 		rec := statsRecord{
-			Ts:       time.Unix(s.Time.Sec, s.Time.Nsec).UTC().Format(time.RFC3339Nano),
+			Ts:       wallTime(s),
 			Kind:     "raw",
 			Counters: countersMap(s),
 		}
@@ -128,7 +128,7 @@ func emitStatsDiff(out io.Writer, s, last cpworker.StatsSummary, format string)
 	switch format {
 	case FormatJSONL:
 		rec := statsRecord{
-			Ts:        t1.UTC().Format(time.RFC3339Nano),
+			Ts:        wallTime(s),
 			Kind:      "rate",
 			IntervalS: &secs,
 			Counters:  countersMap(s),
@@ -142,6 +142,11 @@ func emitStatsDiff(out io.Writer, s, last cpworker.StatsSummary, format string)
 	}
 }
 
+// wallTime formats when cpworker took the snapshot.
+func wallTime(s cpworker.StatsSummary) string {
+	return time.Unix(s.WallTime.Sec, s.WallTime.Nsec).UTC().Format(time.RFC3339Nano)
+}
+
 func writeJSONL(out io.Writer, v any) error {
 	data, err := json.Marshal(v)
 	if err != nil {
@@ -171,6 +176,7 @@ func countersMap(s cpworker.StatsSummary) map[string]any {
 		"error_drop_packets":     packetsAsJSON(s.Output.ErrorDropPackets),
 		"ratelimit_drop_bytes":   bytesAsJSON(s.Output.RatelimitDropBytes),
 		"ratelimit_drop_packets": packetsAsJSON(s.Output.RatelimitDropPackets),
+		"heartbeat_packets":      packetsAsJSON(s.Output.HeartbeatPackets),
 	}
 }
 
@@ -205,6 +211,7 @@ func ratesMap(s, last cpworker.StatsSummary, secs float64) map[string]any {
 	addPacketsRate("error_drop_packets_per_sec", s.Output.ErrorDropPackets, last.Output.ErrorDropPackets)
 	addBytesRate("ratelimit_drop_bytes_per_sec", s.Output.RatelimitDropBytes, last.Output.RatelimitDropBytes)
 	addPacketsRate("ratelimit_drop_packets_per_sec", s.Output.RatelimitDropPackets, last.Output.RatelimitDropPackets)
+	addPacketsRate("heartbeat_packets_per_sec", s.Output.HeartbeatPackets, last.Output.HeartbeatPackets)
 	return rates
 }
 
@@ -233,6 +240,7 @@ func printRawText(out io.Writer, s cpworker.StatsSummary) {
 		{"Error Drop Packets", formatPacketsStats(s.Output.ErrorDropPackets)},
 		{"Ratelimit Drop Bytes", formatBytesStats(s.Output.RatelimitDropBytes)},
 		{"Ratelimit Drop Packets", formatPacketsStats(s.Output.RatelimitDropPackets)},
+		{"Heartbeat Packets", formatPacketsStats(s.Output.HeartbeatPackets)},
 	}
 	maxLen := 0
 	for _, r := range rows {
@@ -246,8 +254,8 @@ func printRawText(out io.Writer, s cpworker.StatsSummary) {
 }
 
 func printSummaryStats(out io.Writer, stats cpworker.StatsSummary, lastStats cpworker.StatsSummary) {
-	headers := make([]string, 0, 12)
-	row := make([]string, 0, 12)
+	headers := make([]string, 0, 13)
+	row := make([]string, 0, 13)
 
 	t1 := time.Unix(stats.Time.Sec, stats.Time.Nsec)
 	t2 := time.Unix(lastStats.Time.Sec, lastStats.Time.Nsec)
@@ -284,6 +292,7 @@ func printSummaryStats(out io.Writer, stats cpworker.StatsSummary, lastStats cpw
 	addPackets("Error Drop Packets", stats.Output.ErrorDropPackets, lastStats.Output.ErrorDropPackets)
 	addBytes("Ratelimit Drop Bytes", stats.Output.RatelimitDropBytes, lastStats.Output.RatelimitDropBytes)
 	addPackets("Ratelimit Drop Packets", stats.Output.RatelimitDropPackets, lastStats.Output.RatelimitDropPackets)
+	addPackets("Heartbeat Packets", stats.Output.HeartbeatPackets, lastStats.Output.HeartbeatPackets)
 
 	maxHeaderLen := 0
 	for _, h := range headers {
diff --git a/cpctl/cmd/stats_test.go b/cpctl/cmd/stats_test.go
index 48223dfc..28ea61c0 100644
--- a/cpctl/cmd/stats_test.go
+++ b/cpctl/cmd/stats_test.go
@@ -164,3 +164,82 @@ func TestPacketsStatsPerSec_ZeroSecs(t *testing.T) {
 		t.Errorf("zero seconds should return input unchanged, got %+v", got)
 	}
 }
+
+// #297: ts is the snapshot's wall-clock time, not its CLOCK_MONOTONIC time.
+func TestRunStats_TsIsSnapshotWallTime(t *testing.T) {
+	withWall := func(s cpworker.StatsSummary, sec int64) cpworker.StatsSummary {
+		s.WallTime.Sec = sec
+		s.WallTime.Nsec = 250000000
+		return s
+	}
+	cases := []struct {
+		name      string
+		count     int
+		snapshots []cpworker.StatsSummary
+		wantTs    string
+	}{
+		{"raw", 1, []cpworker.StatsSummary{withWall(makeStats(632655, 0, 0, 0, 0), 1759900000)}, "2025-10-08T05:06:40.25Z"},
+		{"rate", 2, []cpworker.StatsSummary{
+			withWall(makeStats(632655, 0, 0, 0, 0), 1759900000),
+			withWall(makeStats(632660, 0, 0, 0, 0), 1759900005),
+		}, "2025-10-08T05:06:45.25Z"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := &fakeStatsClient{snapshots: tc.snapshots}
+			var buf bytes.Buffer
+			if err := runStats(context.Background(), client, &buf, tc.count, time.Microsecond, FormatJSONL); err != nil {
+				t.Fatal(err)
+			}
+			var rec statsRecord
+			if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+				t.Fatalf("decode: %v\n%s", err, buf.String())
+			}
+			if rec.Ts != tc.wantTs {
+				t.Errorf("ts = %q, want %q", rec.Ts, tc.wantTs)
+			}
+		})
+	}
+}
+
+// #299: heartbeats are reported in their own counter.
+func TestRunStats_ReportsHeartbeatPackets(t *testing.T) {
+	s0 := makeStats(1000, 0, 0, 0, 0)
+	s1 := makeStats(1005, 0, 0, 0, 0)
+	s1.Output.HeartbeatPackets = cpworker.PacketsStats{Packets: 5}
+
+	t.Run("jsonl", func(t *testing.T) {
+		client := &fakeStatsClient{snapshots: []cpworker.StatsSummary{s0, s1}}
+		var buf bytes.Buffer
+		if err := runStats(context.Background(), client, &buf, 2, time.Microsecond, FormatJSONL); err != nil {
+			t.Fatal(err)
+		}
+		var rec statsRecord
+		if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		hb, ok := rec.Counters["heartbeat_packets"].(map[string]any)
+		if !ok || hb["packets"] != float64(5) {
+			t.Errorf("counters.heartbeat_packets = %v, want packets 5", rec.Counters["heartbeat_packets"])
+		}
+		rate, ok := rec.Rates["heartbeat_packets_per_sec"].(map[string]any)
+		if !ok || rate["packets"] != float64(1) {
+			t.Errorf("rates.heartbeat_packets_per_sec = %v, want packets 1", rec.Rates["heartbeat_packets_per_sec"])
+		}
+	})
+	t.Run("text", func(t *testing.T) {
+		for _, count := range []int{1, 2} {
+			client := &fakeStatsClient{snapshots: []cpworker.StatsSummary{s1, s1}}
+			if count == 2 {
+				client.snapshots = []cpworker.StatsSummary{s0, s1}
+			}
+			var buf bytes.Buffer
+			if err := runStats(context.Background(), client, &buf, count, time.Microsecond, FormatText); err != nil {
+				t.Fatal(err)
+			}
+			if !strings.Contains(buf.String(), "Heartbeat Packets") {
+				t.Errorf("-n %d text output has no Heartbeat Packets row:\n%s", count, buf.String())
+			}
+		}
+	})
+}
diff --git a/cpgolib/cpworker/stats.go b/cpgolib/cpworker/stats.go
index 34216f26..57f2ac29 100644
--- a/cpgolib/cpworker/stats.go
+++ b/cpgolib/cpworker/stats.go
@@ -10,10 +10,17 @@ const (
 )
 
 type StatsSummary struct {
+	// Time is the snapshot's CLOCK_MONOTONIC time. Use it for the interval
+	// between two snapshots; it is not a date.
 	Time struct {
 		Sec  int64 `mapstructure:"sec"`
 		Nsec int64 `mapstructure:"nsec"`
 	} `mapstructure:"time"`
+	// WallTime is the snapshot's CLOCK_REALTIME time: when it was taken.
+	WallTime struct {
+		Sec  int64 `mapstructure:"sec"`
+		Nsec int64 `mapstructure:"nsec"`
+	} `mapstructure:"wall_time"`
 	Capture        CaptureStats        `mapstructure:"capture"`
 	Output         OutputStats         `mapstructure:"output"`
 	PipelineBuffer PipelineBufferStats `mapstructure:"pipeline_buffer"`
diff --git a/cpgolib/cpworker/unix_client_test.go b/cpgolib/cpworker/unix_client_test.go
index c0f2d70a..4d7cb839 100644
--- a/cpgolib/cpworker/unix_client_test.go
+++ b/cpgolib/cpworker/unix_client_test.go
@@ -209,6 +209,35 @@ func TestUnixClient_Info(t *testing.T) {
 	}
 }
 
+func TestUnixClient_CollectStatsSummaryTimes(t *testing.T) {
+	srv := newFakeServer(t)
+	defer srv.Close()
+
+	srv.handlers["collect_stats_summary"] = func(args map[string]any) (map[string]any, error) {
+		return map[string]any{
+			"time":      map[string]any{"sec": 632655, "nsec": 154030865},
+			"wall_time": map[string]any{"sec": 1759900000, "nsec": 250000000},
+		}, nil
+	}
+
+	client, err := NewClient(srv.connStr())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	stats, err := client.CollectStatsSummary(context.Background())
+	if err != nil {
+		t.Fatalf("collect_stats_summary: %v", err)
+	}
+	if stats.Time.Sec != 632655 || stats.Time.Nsec != 154030865 {
+		t.Errorf("time: got %+v", stats.Time)
+	}
+	if stats.WallTime.Sec != 1759900000 || stats.WallTime.Nsec != 250000000 {
+		t.Errorf("wall_time: got %+v", stats.WallTime)
+	}
+}
+
 func TestUnixClient_UnknownCommand(t *testing.T) {
 	srv := newFakeServer(t)
 	defer srv.Close()
diff --git a/cpworker/src/output_zmq.c b/cpworker/src/output_zmq.c
index 94711c39..4beda450 100644
--- a/cpworker/src/output_zmq.c
+++ b/cpworker/src/output_zmq.c
@@ -113,11 +113,17 @@ int zmq_flush_packet(zmq_output_t *output)
         output->error_info.first_pktsec = pkts_buf->first_pktsec;
     }
 
+    // Only captured packets are counted as forwarded or dropped. A batch that carries
+    // nothing but heartbeats contributes no bytes, not even its batch header.
+    uint16_t data_num = send_num - pkts_buf->heartbeat_num;
+    uint32_t data_bytes = data_num > 0 ? pkts_buf->batch_bufpos - pkts_buf->heartbeat_bytes : 0;
+
     int rc = zmq_send(output->pusher, &(pkts_buf->buf[0]), pkts_buf->batch_bufpos, ZMQ_DONTWAIT);
     if (rc != -1)
     {
-        bytes_stats_add(&output->base.stats->fwd_bytes, pkts_buf->batch_bufpos);
-        packets_stats_add(&output->base.stats->fwd_packets, send_num);
+        bytes_stats_add(&output->base.stats->fwd_bytes, data_bytes);
+        packets_stats_add(&output->base.stats->fwd_packets, data_num);
+        packets_stats_add(&output->base.stats->heartbeat_packets, pkts_buf->heartbeat_num);
     }
     else
     {
@@ -125,15 +131,17 @@ int zmq_flush_packet(zmq_output_t *output)
             snprintf(output->error_info.send_error, ERROR_BUFFER_SIZE, "zmq_send failed: %s", zmq_strerror(errno));
 
         output->error_info.nb_drop_batches++;
-        output->error_info.nb_drop_packets += send_num;
+        output->error_info.nb_drop_packets += data_num;
 
-        bytes_stats_add(&output->base.stats->error_drop_bytes, pkts_buf->batch_bufpos);
-        packets_stats_add(&output->base.stats->error_drop_packets, send_num);
+        bytes_stats_add(&output->base.stats->error_drop_bytes, data_bytes);
+        packets_stats_add(&output->base.stats->error_drop_packets, data_num);
     }
 
     pkts_buf->first_pktsec = 0;
     pkts_buf->batch_bufpos = sizeof(zmq_pkt_batch_hdr_t);
     pkts_buf->batch_hdr.pkts_num = 0;
+    pkts_buf->heartbeat_num = 0;
+    pkts_buf->heartbeat_bytes = 0;
     return 0;
 }
 
@@ -336,14 +344,15 @@ static void zmq_send_heartbeat_packet(zmq_output_t *output, struct timeval *tv)
     memcpy(&pkts_buf->buf[pos + offsetof(struct ether_header, ether_type)], &hb_etype, sizeof(hb_etype));
     pos += sizeof(eth);
 
+    pkts_buf->heartbeat_bytes += pos - pkts_buf->batch_bufpos;
+    pkts_buf->heartbeat_num++;
     pkts_buf->batch_bufpos = pos;
     pkts_buf->batch_hdr.pkts_num++;
 
-    // Flush immediately - heartbeat triggers batch flush
+    // Flush immediately - heartbeat triggers batch flush. The flush counts the heartbeat
+    // in heartbeat_packets once it is sent.
     zmq_flush_packet(output);
 
-    // Update heartbeat stats and last packet time
-    packets_stats_add(&output->base.stats->heartbeat_packets, 1);
     output->last_pkt_tv = *tv;
 }
 
diff --git a/cpworker/src/output_zmq.h b/cpworker/src/output_zmq.h
index 4ccffb54..68fd7b5f 100644
--- a/cpworker/src/output_zmq.h
+++ b/cpworker/src/output_zmq.h
@@ -43,6 +43,10 @@ typedef struct
     char buf[ZMQ_MAX_BATCH_BUF_SIZE];
     uint32_t batch_bufpos;
     long int first_pktsec;
+    // Heartbeat records in the batch. They are not captured packets, so a flush keeps
+    // them out of the fwd/error_drop counters and counts them in heartbeat_packets.
+    uint16_t heartbeat_num;
+    uint32_t heartbeat_bytes;
 } zmq_pkts_buf_t;
 
 typedef struct ZmqOptions
diff --git a/cpworker/src/task.c b/cpworker/src/task.c
index 03c68095..332f6c35 100644
--- a/cpworker/src/task.c
+++ b/cpworker/src/task.c
@@ -157,7 +157,8 @@ static task_error_t *task_error_new(int index, const char *fingerprint, const ch
 
 typedef struct TaskStatsSummarySnapshot
 {
-    struct timespec tm;
+    struct timespec tm;      // CLOCK_MONOTONIC: interval between two snapshots
+    struct timespec wall_tm; // CLOCK_REALTIME: when the snapshot was taken
     capture_stats_t capture;
     output_stats_t output;
     pipeline_buffer_stats_t pipeline_buffer;
@@ -644,6 +645,7 @@ static void task_manager_update_stats_summary()
     memset(&stats, 0, sizeof(task_stats_summary_snapshot_t));
 
     clock_gettime(CLOCK_MONOTONIC, &stats.tm);
+    clock_gettime(CLOCK_REALTIME, &stats.wall_tm);
     stats.capture = this->stats_summary.capture;
     stats.output = this->stats_summary.output;
     if (this->execution_model == EXECUTION_MODEL_PIPELINE)
@@ -731,6 +733,11 @@ int task_manager_collect_stats_summary_command(cJSON *cmd_msg, cJSON *server_msg
         !cJSON_AddNumberToObject(time, "nsec", stats.tm.tv_nsec))
         return -1;
 
+    cJSON *wall_time = cJSON_AddObjectToObject(server_msg, "wall_time");
+    if (!wall_time || !cJSON_AddNumberToObject(wall_time, "sec", stats.wall_tm.tv_sec) ||
+        !cJSON_AddNumberToObject(wall_time, "nsec", stats.wall_tm.tv_nsec))
+        return -1;
+
     if (add_capture_stats(server_msg, "capture", &stats.capture) != 0 ||
         add_output_stats(server_msg, "output", &stats.output) != 0 ||
         add_pipeline_buffer_stats(server_msg, "pipeline_buffer", &stats.pipeline_buffer) != 0)
diff --git a/cpworker/tests/unit/misc.c b/cpworker/tests/unit/misc.c
index 9589d0d0..ef51ddbe 100644
--- a/cpworker/tests/unit/misc.c
+++ b/cpworker/tests/unit/misc.c
@@ -315,13 +315,13 @@ void test_zmq_heartbeat_packet_generation(void)
     // Call heartbeat - should generate a heartbeat packet
     output_heartbeat((output_base_t *)output, now_tv.tv_sec);
 
-    // Verify heartbeat stats were incremented
+    // The heartbeat is queued on the not-yet-connected pipe, so the send succeeds
     TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
 
-    // The ZMQ send may fail (no receiver) but the packet was still generated
-    // Either forwarded or error-dropped (depends on ZMQ connection state)
-    uint64_t total = stats.fwd_packets.packets + stats.error_drop_packets.packets;
-    TEST_ASSERT_EQUAL_UINT64(1, total);
+    // A heartbeat is not a captured packet: it is neither forwarded nor dropped
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_bytes.bytes);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.error_drop_packets.packets);
 
     zmq_output_destroy((output_base_t *)output);
 }
diff --git a/cpworker/tests/unit/output_zmq.c b/cpworker/tests/unit/output_zmq.c
index f537df97..04e5836b 100644
--- a/cpworker/tests/unit/output_zmq.c
+++ b/cpworker/tests/unit/output_zmq.c
@@ -3,6 +3,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <time.h>
 
 #include <zmq.h>
 
@@ -13,6 +14,7 @@
 #include "pkt_dir.h"
 
 extern int zmq_flush_packet(zmq_output_t *output);
+extern void zmq_heartbeat(output_base_t *self, time_t now);
 
 #define BATCH_HDR_SIZE sizeof(zmq_pkt_batch_hdr_t)
 #define RECORD_HDR_SIZE (ZMQ_PKT_DATA_LEN_SIZE + sizeof(zmq_pkt_hdr_t))
@@ -330,6 +332,137 @@ void test_runt_and_normal_frame_both_accounted(void)
     TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets);
 }
 
+/* ---- #299: heartbeats are not forwarded packets ---- */
+
+#define HEARTBEAT_RECORD_SIZE (RECORD_HDR_SIZE + 14)
+
+static zmq_output_t *new_heartbeat_output(uint16_t port, int hwm, char *errbuf)
+{
+    zmq_options_t opts = {
+        .host = "127.0.0.1",
+        .port = port,
+        .hwm = hwm,
+        .service_tag = 1,
+        .uuid = (char *)VALID_UUID,
+        .rate_limit_mbps = 0,
+        .slice = 0,
+        .heartbeat_ms = 1000,
+    };
+    return zmq_output_new(opts, &stats, errbuf);
+}
+
+// Makes the last packet one second old, so the next zmq_heartbeat() sends a heartbeat
+// without flushing a batch whose first packet is from that same second.
+static time_t make_heartbeat_due(zmq_output_t *output)
+{
+    time_t now = time(NULL);
+    output->last_pkt_tv.tv_sec = now - 1;
+    output->last_pkt_tv.tv_usec = 0;
+    return now;
+}
+
+static void send_frame_at(zmq_output_t *output, time_t sec)
+{
+    uint8_t frame[64];
+    memset(frame, 0x5a, sizeof(frame));
+    frame[12] = 0x08;
+    frame[13] = 0x00;
+
+    struct pcap_pkthdr hdr;
+    memset(&hdr, 0, sizeof(hdr));
+    hdr.ts.tv_sec = sec;
+    hdr.caplen = sizeof(frame);
+    hdr.len = sizeof(frame);
+    TEST_ASSERT_EQUAL_INT(0, output_send_packet(&output->base, &hdr, frame, PKT_DIR_INCOMING));
+}
+
+static void destroy_without_linger(zmq_output_t *output)
+{
+    int linger = 0;
+    zmq_setsockopt(output->pusher, ZMQ_LINGER, &linger, sizeof(linger));
+    zmq_output_destroy(&output->base);
+}
+
+void test_heartbeat_on_idle_output_is_not_forwarded(void)
+{
+    char errbuf[ERROR_BUFFER_SIZE] = {0};
+    zmq_output_t *output = new_heartbeat_output(receiver_port, 100, errbuf);
+    TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf);
+
+    zmq_heartbeat(&output->base, make_heartbeat_due(output));
+
+    TEST_ASSERT_EQUAL_INT(BATCH_HDR_SIZE + HEARTBEAT_RECORD_SIZE, recv_batch());
+    TEST_ASSERT_EQUAL_UINT16(1, batch_pkts_num());
+    TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_bytes.bytes);
+
+    zmq_output_destroy(&output->base);
+}
+
+void test_heartbeat_in_batch_with_data_counts_only_the_data(void)
+{
+    char errbuf[ERROR_BUFFER_SIZE] = {0};
+    zmq_output_t *output = new_heartbeat_output(receiver_port, 100, errbuf);
+    TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf);
+
+    time_t now = time(NULL);
+    send_frame_at(output, now - 1);
+    zmq_heartbeat(&output->base, make_heartbeat_due(output));
+
+    int n = recv_batch();
+    TEST_ASSERT_GREATER_THAN_INT(0, n);
+    TEST_ASSERT_EQUAL_UINT16(2, batch_pkts_num());
+    TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(1, stats.fwd_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64((uint64_t)n - HEARTBEAT_RECORD_SIZE, stats.fwd_bytes.bytes);
+
+    zmq_output_destroy(&output->base);
+}
+
+void test_failed_heartbeat_is_not_counted(void)
+{
+    // Nothing listens on port 1. With hwm 1 the unconnected pipe holds one batch, so the
+    // first heartbeat is queued and the second send fails.
+    char errbuf[ERROR_BUFFER_SIZE] = {0};
+    zmq_output_t *output = new_heartbeat_output(1, 1, errbuf);
+    TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf);
+
+    zmq_heartbeat(&output->base, make_heartbeat_due(output));
+    TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
+
+    zmq_heartbeat(&output->base, make_heartbeat_due(output));
+    TEST_ASSERT_EQUAL_UINT64(1, output->error_info.nb_drop_batches);
+    TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.error_drop_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.error_drop_bytes.bytes);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets);
+
+    destroy_without_linger(output);
+}
+
+void test_failed_batch_with_heartbeat_drops_only_the_data(void)
+{
+    char errbuf[ERROR_BUFFER_SIZE] = {0};
+    zmq_output_t *output = new_heartbeat_output(1, 1, errbuf);
+    TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf);
+
+    zmq_heartbeat(&output->base, make_heartbeat_due(output)); // fills the pipe
+
+    time_t now = time(NULL);
+    send_frame_at(output, now - 1);
+    uint32_t batch_size = output->pkts_buf.batch_bufpos + HEARTBEAT_RECORD_SIZE;
+    zmq_heartbeat(&output->base, make_heartbeat_due(output));
+
+    TEST_ASSERT_EQUAL_UINT64(1, output->error_info.nb_drop_batches);
+    TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(batch_size - HEARTBEAT_RECORD_SIZE, stats.error_drop_bytes.bytes);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets);
+
+    destroy_without_linger(output);
+}
+
 int main(void)
 {
     UNITY_BEGIN();
@@ -350,5 +483,10 @@ int main(void)
     RUN_TEST(test_vlan_stack_running_past_caplen_is_error_drop);
     RUN_TEST(test_runt_and_normal_frame_both_accounted);
 
+    RUN_TEST(test_heartbeat_on_idle_output_is_not_forwarded);
+    RUN_TEST(test_heartbeat_in_batch_with_data_counts_only_the_data);
+    RUN_TEST(test_failed_heartbeat_is_not_counted);
+    RUN_TEST(test_failed_batch_with_heartbeat_drops_only_the_data);
+
     return UNITY_END();
 }
diff --git a/cpworker/tests/unit/unix_control.c b/cpworker/tests/unit/unix_control.c
index 94d7fe69..5fe53780 100644
--- a/cpworker/tests/unit/unix_control.c
+++ b/cpworker/tests/unit/unix_control.c
@@ -2,6 +2,7 @@
  *   - unix_rpc_basic: ping / info command handlers
  *   - unix-manager:   SO_SNDTIMEO hardening against a slow reader stalling
  *                     the manager thread in send()
+ *   - task:           collect_stats_summary snapshot timestamps
  *
  * Merged into one test binary so the cpworker sources only get compiled once
  * per test run instead of per-file. */
@@ -18,6 +19,7 @@
 #include "unity/src/unity.h"
 
 #include "cJSON/cJSON.h"
+#include "task.h"
 #include "unix-manager.h"
 #include "unix_rpc_basic.h"
 
@@ -199,6 +201,43 @@ void test_send_timeout_fires_on_unread_peer(void)
     close(fds[1]);
 }
 
+/* ---------- #297: stats snapshot carries a wall-clock time ---------- */
+
+static int64_t json_time_sec(cJSON *resp, const char *key)
+{
+    cJSON *obj = cJSON_GetObjectItemCaseSensitive(resp, key);
+    TEST_ASSERT_TRUE_MESSAGE(cJSON_IsObject(obj), key);
+    cJSON *sec = cJSON_GetObjectItemCaseSensitive(obj, "sec");
+    cJSON *nsec = cJSON_GetObjectItemCaseSensitive(obj, "nsec");
+    TEST_ASSERT_TRUE(cJSON_IsNumber(sec));
+    TEST_ASSERT_TRUE(cJSON_IsNumber(nsec));
+    return (int64_t)sec->valuedouble;
+}
+
+void test_stats_summary_has_wall_time_and_monotonic_time(void)
+{
+    struct timespec mono_before, mono_after;
+    clock_gettime(CLOCK_MONOTONIC, &mono_before);
+    time_t wall_before = time(NULL);
+    task_manager_update_stats();
+    time_t wall_after = time(NULL);
+    clock_gettime(CLOCK_MONOTONIC, &mono_after);
+
+    cJSON *resp = cJSON_CreateObject();
+    TEST_ASSERT_NOT_NULL(resp);
+    TEST_ASSERT_EQUAL(0, task_manager_collect_stats_summary_command(NULL, resp, NULL));
+
+    int64_t wall = json_time_sec(resp, "wall_time");
+    TEST_ASSERT_GREATER_OR_EQUAL_INT64((int64_t)wall_before, wall);
+    TEST_ASSERT_LESS_OR_EQUAL_INT64((int64_t)wall_after, wall);
+
+    int64_t mono = json_time_sec(resp, "time");
+    TEST_ASSERT_GREATER_OR_EQUAL_INT64((int64_t)mono_before.tv_sec, mono);
+    TEST_ASSERT_LESS_OR_EQUAL_INT64((int64_t)mono_after.tv_sec, mono);
+
+    cJSON_Delete(resp);
+}
+
 int main(void)
 {
     UNITY_BEGIN();
@@ -208,5 +247,6 @@ int main(void)
     RUN_TEST(test_info_command_working_dir_updates_on_reset);
     RUN_TEST(test_set_send_timeout_readable_via_getsockopt);
     RUN_TEST(test_send_timeout_fires_on_unread_peer);
+    RUN_TEST(test_stats_summary_has_wall_time_and_monotonic_time);
     return UNITY_END();
 }
diff --git a/docs/REFACTOR-CPCTL.md b/docs/REFACTOR-CPCTL.md
index f3601983..24415b62 100644
--- a/docs/REFACTOR-CPCTL.md
+++ b/docs/REFACTOR-CPCTL.md
@@ -138,9 +138,9 @@ Notes:
 |---|---|---|---|
 | `--unix` | `-u` | unix socket path | raw path, no `unix://` prefix |
 | `--count` | `-n` | sample count, 0 = forever | matches `top -n`, `head -n` |
-| `--interval` | `-i` | poll interval | matches `ping -i` |
+| `--interval` | `-i` | poll interval | name from `ping -i`; value is a Go duration with a unit (`5s`, `500ms`) |
 | `--format` | `-f` | `text\|jsonl` | default `text` |
-| `--timeout` | `-W` | per-RPC timeout | matches `ping -W` |
+| `--timeout` | `-W` | per-RPC timeout | name from `ping -W`; value is a Go duration with a unit (`3s`) |
 | `--quiet` | `-q` | (ping only) suppress per-tick output, only summary | matches `ping -q` |
 
 ### Why `-n` for count, not `-c`
@@ -324,7 +324,7 @@ available:
 
 ```json
 {"ts":"...","kind":"raw","interval_sec":null,"counters":{...},"rates":null}
-{"ts":"...","kind":"rate","interval_sec":2,"counters":{...},"rates":{"cap_bytes_per_sec":{...},...}}
+{"ts":"...","kind":"rate","interval_sec":5,"counters":{...},"rates":{"cap_bytes_per_sec":{...},...}}
 ```
 
 `-n 1` emits one `raw` line. `-n >= 2` and `-n 0` emit only `rate` lines; the
@@ -332,6 +332,19 @@ first poll is the baseline and is not printed. A single rate is `null` when
 its counter went backwards between two samples (e.g. across a cpworker
 restart).
 
+Sampling and timestamps:
+
+- cpworker refreshes the counters that `collect_stats_summary` returns once
+  every 5 s. A poll that returns a snapshot cpctl already has emits no record.
+- `interval_sec` is the time between the two snapshots, taken from their
+  `CLOCK_MONOTONIC` times (`time` in the RPC reply), and the rates are
+  computed over it. It can be longer than `--interval`.
+- `ts` is the wall-clock time at which cpworker took the snapshot
+  (`wall_time` in the RPC reply). With `-n 1` the counters can be up to 5 s
+  old.
+- `counters.heartbeat_packets` counts the zmq heartbeats sent. Heartbeats are
+  not captured packets and are not part of `fwd_packets` / `fwd_bytes`.
+
 ### 6.3 info
 
 Minimum useful set, all derivable from data cpworker already has:
@@ -437,7 +450,7 @@ service manager handles aggregation. `cpctl info` will surface
   `ping -i`.** No single tool uses both letters with these meanings —
   `top` uses `-d` (delay) for interval, `ping` uses `-c` for count — so we
   borrow each from the closer analog: `top` for iterative dashboards,
-  `ping` for latency. `cpctl stats -n 5 -i 0.5` reads cleanly under either
+  `ping` for latency. `cpctl stats -n 5 -i 10s` reads cleanly under either
   mental model.
 - **`info` is the discovery anchor.** An agent calling `cpctl info -f jsonl`
   gets enough to bootstrap: which config is loaded, where logs go, how long
diff --git a/docs/ZMQ-WIRE-FORMAT-zh-Hans.md b/docs/ZMQ-WIRE-FORMAT-zh-Hans.md
index b82600a3..74a13378 100644
--- a/docs/ZMQ-WIRE-FORMAT-zh-Hans.md
+++ b/docs/ZMQ-WIRE-FORMAT-zh-Hans.md
@@ -210,7 +210,16 @@ EtherType 为 **`0xFFFF`**（`ZMQ_HEARTBEAT_ETHER_TYPE`）。没有 MPLS 标签
 | 生成 heartbeat | — |
 | `zmq_send()` 会被阻塞时跳过 | `ZMQ_DONTWAIT` — 该 batch 记为丢弃 |
 
-发送使用 `ZMQ_DONTWAIT`；失败时整批丢弃，计入 `error_drop_batches` / `error_drop_packets`。
+发送使用 `ZMQ_DONTWAIT`；失败时整批丢弃。
+
+heartbeat 记录不是抓到的报文，输出计数器不计入它：
+
+| batch 内容 | 发送成功 | 发送失败 |
+| --- | --- | --- |
+| 有抓到的报文（可能夹带 heartbeat） | 数据记录和 batch header → `fwd_packets` / `fwd_bytes`；heartbeat → `heartbeat_packets` | 数据记录和 batch header → `error_drop_packets` / `error_drop_bytes` |
+| 只有 heartbeat | `heartbeat_packets` | 不计数 |
+
+字节数为 batch 字节数减去 heartbeat 记录。丢弃的 batch 个数只出现在 worker 的周期性错误日志里（`nb_drop_batches`），没有对应的统计计数器。
 
 ---
 
diff --git a/docs/ZMQ-WIRE-FORMAT.md b/docs/ZMQ-WIRE-FORMAT.md
index 800ed501..55311a9a 100644
--- a/docs/ZMQ-WIRE-FORMAT.md
+++ b/docs/ZMQ-WIRE-FORMAT.md
@@ -229,8 +229,17 @@ happens:
 | A heartbeat is generated | — |
 | `zmq_send()` would block and is skipped | `ZMQ_DONTWAIT` — the batch is counted as dropped |
 
-The send uses `ZMQ_DONTWAIT`; on failure the batch is dropped and counted in
-`error_drop_batches` / `error_drop_packets`.
+The send uses `ZMQ_DONTWAIT`; on failure the whole batch is dropped.
+
+Heartbeat records are not captured packets, so the output counters leave them out:
+
+| Batch | Sent | Dropped |
+| --- | --- | --- |
+| Captured packets (and heartbeats, if any) | data records and batch header → `fwd_packets` / `fwd_bytes`; heartbeats → `heartbeat_packets` | data records and batch header → `error_drop_packets` / `error_drop_bytes` |
+| Heartbeats only | `heartbeat_packets` | not counted |
+
+Bytes are the batch bytes minus the heartbeat records. The worker's periodic error log reports
+the number of dropped batches (`nb_drop_batches`); no stats counter carries it.
 
 ---