diff --git a/README.md b/README.md
index 366b1f5..00eb4fd 100644
--- a/README.md
+++ b/README.md
@@ -116,6 +116,13 @@ const result = engine.evaluate(.log, &my_log_accessor, &my_log_ctx, &policy_id_b
     .io = io,
 });
 
+// Volume (spec v1.7.1): every evaluate call is counted regardless of match and
+// reported on the next sync, which resets the counters whether or not that sync
+// succeeds — reported volume is a lower bound, never replayed. Byte volume is
+// opt-in: pass the batch's uncompressed OTLP protobuf size, which only the
+// caller can measure.
+registry.volume.addBytes(.log, @intCast(request_size_bytes));
+
 // Transforms whose required primitive (set/delete/move) is unwired on the
 // accessor are *eliminated at compile time* for that callsite — no
 // runtime branch, no code emitted. Different consumers can apply
diff --git a/proto/tero/policy/v1/policy.proto b/proto/tero/policy/v1/policy.proto
index b1ddbca..e2a3ee5 100644
--- a/proto/tero/policy/v1/policy.proto
+++ b/proto/tero/policy/v1/policy.proto
@@ -126,6 +126,40 @@ message PolicySyncStatus {
   TransformStageStatus add = 13;
 }
 
+// VolumeStats reports the total telemetry a client observed since the last
+// sync, regardless of whether any policy matched it. Counts are of records
+// entering policy evaluation, before any keep or transform stage runs.
+//
+// Counters are reset when they are read into a sync request, whether or not
+// that sync then succeeds — the same rule PolicySyncStatus.match_hits and
+// match_misses follow. A failed sync loses its interval from the numerator and
+// the denominator alike, so match rates stay meaningful; counters from a failed
+// sync must never be replayed, since the server cannot tell a replay from new
+// telemetry. Reported volume is a lower bound, not an exact total.
+//
+// Reporting volume is optional, and every field is individually optional: an
+// implementation may report record counts without byte counts, or a subset of
+// signals. Any field left at 0 means "not tracked" as much as it means "none
+// seen", so consumers must not read 0 as an observation.
+//
+// Byte counts, when reported, are the uncompressed OTLP protobuf serialized
+// size of the records as received, and are an estimate; implementations that
+// cannot measure this cheaply may approximate it. A size in any other encoding
+// must not be reported here — leave the field at 0 instead.
+message VolumeStats {
+  // Log records seen, and their total size in bytes.
+  int64 log_records = 1;
+  int64 log_bytes = 2;
+
+  // Metric data points seen, and their total size in bytes.
+  int64 metric_data_points = 3;
+  int64 metric_bytes = 4;
+
+  // Spans seen, and their total size in bytes.
+  int64 spans = 5;
+  int64 span_bytes = 6;
+}
+
 // SyncRequest is sent by clients to request policy updates.
 message SyncRequest {
   // Client identification and capabilities
@@ -142,6 +176,10 @@ message SyncRequest {
 
   // Status of individual policies within this set.
   repeated PolicySyncStatus policy_statuses = 5;
+
+  // Optional. Total telemetry observed since the last sync, regardless of
+  // policy match. Clients that do not track volume omit this.
+  VolumeStats volume = 6;
 }
 
 enum SyncType {
diff --git a/src/policy/policy_engine.zig b/src/policy/policy_engine.zig
index 8e33845..25b2a1d 100644
--- a/src/policy/policy_engine.zig
+++ b/src/policy/policy_engine.zig
@@ -336,6 +336,12 @@ pub const PolicyEngine = struct {
         policy_id_buf: [][]const u8,
         options: EvaluateOptions,
     ) PolicyResult {
+        // Volume tracking (v1.7.1): every record entering evaluation counts,
+        // regardless of match — including records with no policies loaded for
+        // their signal, so this must precede both early returns below. Byte
+        // volume is opt-in via `registry.volume.addBytes`.
+        self.registry.volume.record(T);
+
         // Get current snapshot from registry (lock-free)
         const snapshot = self.registry.getSnapshot() orelse {
             const event: EvaluateEmpty = .{};
@@ -973,6 +979,105 @@ test "PolicyEngine: empty registry returns unset" {
     try testing.expectEqual(@as(usize, 0), result.matched_policy_ids.len);
 }
 
+test "PolicyEngine: volume counts every record, including with no policies loaded" {
+    const allocator = testing.allocator;
+
+    var noop_bus: NoopEventBus = undefined;
+    noop_bus.init(std.Options.debug_io);
+    var registry = PolicyRegistry.init(allocator, noop_bus.eventBus());
+    defer registry.deinit();
+
+    const engine = PolicyEngine.init(noop_bus.eventBus(), &registry);
+
+    // No snapshot at all: evaluation returns early, volume still counts.
+    var test_log: TestLogContext = .{ .message = "hello" };
+    var policy_id_buf: [16][]const u8 = undefined;
+    _ = evalTestLog(&engine, &test_log, &policy_id_buf);
+    registry.volume.addBytes(.log, 120);
+
+    // A dropped record counts at its pre-policy size too.
+    var policy: Policy = .{
+        .id = try allocator.dupe(u8, "policy-1"),
+        .enabled = true,
+        .target = .{ .log = .{ .keep = try allocator.dupe(u8, "none") } },
+    };
+    try policy.target.?.log.match.append(allocator, .{
+        .field = .{ .log_field = .LOG_FIELD_BODY },
+        .match = .{ .regex = try allocator.dupe(u8, "error") },
+    });
+    defer policy.deinit(allocator);
+    try registry.updatePolicies(&.{policy}, "file-provider", .file);
+
+    var error_log: TestLogContext = .{ .message = "an error occurred" };
+    const dropped = evalTestLog(&engine, &error_log, &policy_id_buf);
+    try testing.expectEqual(FilterDecision.drop, dropped.decision);
+    registry.volume.addBytes(.log, 80);
+
+    // Bytes are opt-in and independent of record counting.
+    _ = evalTestLog(&engine, &test_log, &policy_id_buf);
+
+    const volume = registry.volume.readAndReset();
+    try testing.expectEqual(@as(i64, 3), volume.log_records);
+    try testing.expectEqual(@as(i64, 200), volume.log_bytes);
+    // Other signals are untracked, not zero-observed — both look like 0 here.
+    try testing.expectEqual(@as(i64, 0), volume.spans);
+    try testing.expectEqual(@as(i64, 0), volume.metric_data_points);
+
+    // Drained: a second read reports nothing.
+    try testing.expect(registry.volume.readAndReset().isZero());
+}
+
+test "PolicyEngine: volume routes each signal to its own counter" {
+    const allocator = testing.allocator;
+
+    var noop_bus: NoopEventBus = undefined;
+    noop_bus.init(std.Options.debug_io);
+    var registry = PolicyRegistry.init(allocator, noop_bus.eventBus());
+    defer registry.deinit();
+
+    // A log-only policy set: the metric and trace indices are empty, so those
+    // evaluations take the `index.isEmpty()` early return and must still count.
+    var policy: Policy = .{
+        .id = try allocator.dupe(u8, "policy-1"),
+        .enabled = true,
+        .target = .{ .log = .{ .keep = try allocator.dupe(u8, "all") } },
+    };
+    try policy.target.?.log.match.append(allocator, .{
+        .field = .{ .log_field = .LOG_FIELD_BODY },
+        .match = .{ .regex = try allocator.dupe(u8, "hello") },
+    });
+    defer policy.deinit(allocator);
+    try registry.updatePolicies(&.{policy}, "file-provider", .file);
+
+    const engine = PolicyEngine.init(noop_bus.eventBus(), &registry);
+    var policy_id_buf: [16][]const u8 = undefined;
+
+    var test_log: TestLogContext = .{ .message = "hello" };
+    _ = evalTestLog(&engine, &test_log, &policy_id_buf);
+
+    var test_metric: TestMetricContext = .{ .name = "http_requests_total" };
+    _ = evalMetric(&engine, &test_metric, &policy_id_buf);
+    _ = evalMetric(&engine, &test_metric, &policy_id_buf);
+
+    var test_span: TestTraceContext = .{ .name = "GET /users" };
+    _ = evalTrace(&engine, &test_span, &policy_id_buf);
+    _ = evalTrace(&engine, &test_span, &policy_id_buf);
+    _ = evalTrace(&engine, &test_span, &policy_id_buf);
+
+    // One counter per signal, no cross-talk between them.
+    const volume = registry.volume.readAndReset();
+    try testing.expectEqual(@as(i64, 1), volume.log_records);
+    try testing.expectEqual(@as(i64, 2), volume.metric_data_points);
+    try testing.expectEqual(@as(i64, 3), volume.spans);
+
+    // Bytes stay untracked unless the consumer opts in, per signal.
+    registry.volume.addBytes(.metric, 42);
+    const bytes = registry.volume.readAndReset();
+    try testing.expectEqual(@as(i64, 42), bytes.metric_bytes);
+    try testing.expectEqual(@as(i64, 0), bytes.log_bytes);
+    try testing.expectEqual(@as(i64, 0), bytes.span_bytes);
+}
+
 test "PolicyEngine: single policy drop match" {
     const allocator = testing.allocator;
 
diff --git a/src/policy/provider.zig b/src/policy/provider.zig
index c0552a0..72397c3 100644
--- a/src/policy/provider.zig
+++ b/src/policy/provider.zig
@@ -19,6 +19,49 @@ pub const PolicyStatsSnapshot = struct {
     errors: []const []const u8 = &.{},
 };
 
+/// Total telemetry that entered policy evaluation since the counters were last
+/// read, regardless of match (spec v1.7.1 `VolumeStats`). Counted before the
+/// keep and transform stages, so dropped/sampled/redacted records are included
+/// at their pre-policy size.
+///
+/// Reported at most once: draining resets, whether or not the sync then
+/// succeeds, so a failed sync loses its interval rather than replaying it. That
+/// keeps volume on the same footing as `hits`/`misses` above — retaining one
+/// side of `(hits + misses) / <signal count>` without the other would skew the
+/// ratio — and a replay would double count, since the server has no interval
+/// identifier to dedupe on. Volume is a lower bound, not an exact total.
+///
+/// Every field is independently optional: `0` means "not tracked" as much as
+/// "none seen". Byte counts are only populated when the consumer calls
+/// `registry.volume.addBytes` — the engine reads records through accessors and
+/// has no serialized form to measure.
+pub const VolumeSnapshot = struct {
+    log_records: i64 = 0,
+    log_bytes: i64 = 0,
+    metric_data_points: i64 = 0,
+    metric_bytes: i64 = 0,
+    spans: i64 = 0,
+    span_bytes: i64 = 0,
+
+    pub fn isZero(self: VolumeSnapshot) bool {
+        inline for (@typeInfo(VolumeSnapshot).@"struct".fields) |f| {
+            if (@field(self, f.name) != 0) return false;
+        }
+        return true;
+    }
+
+    pub fn toProto(self: VolumeSnapshot) proto.policy.VolumeStats {
+        return .{
+            .log_records = self.log_records,
+            .log_bytes = self.log_bytes,
+            .metric_data_points = self.metric_data_points,
+            .metric_bytes = self.metric_bytes,
+            .spans = self.spans,
+            .span_bytes = self.span_bytes,
+        };
+    }
+};
+
 /// Pull-based stats source handed to a provider by the registry at subscribe
 /// time. The provider invokes `collect` immediately before each sync, passing a
 /// per-sync arena; the registry returns one snapshot per policy — including
@@ -28,10 +71,19 @@ pub const PolicyStatsSnapshot = struct {
 pub const StatsCollector = struct {
     context: *anyopaque,
     collect: *const fn (arena: std.mem.Allocator, context: *anyopaque) anyerror![]PolicyStatsSnapshot,
+    /// Drain the registry's volume counters (spec v1.7.1), resetting them.
+    /// Optional: a collector that leaves it null reports no volume, which is
+    /// conformant.
+    collect_volume: ?*const fn (context: *anyopaque) VolumeSnapshot = null,
 
     pub fn call(self: StatsCollector, arena: std.mem.Allocator) anyerror![]PolicyStatsSnapshot {
         return self.collect(arena, self.context);
     }
+
+    pub fn drainVolume(self: StatsCollector) VolumeSnapshot {
+        const f = self.collect_volume orelse return .{};
+        return f(self.context);
+    }
 };
 
 /// Extension sync plumbing (spec v1.6.0), implemented outside policy_zig by
@@ -78,3 +130,60 @@ pub const PolicyCallback = struct {
         try self.onUpdate(self.context, update);
     }
 };
+
+// =============================================================================
+// Tests
+// =============================================================================
+
+const testing = std.testing;
+
+test "VolumeSnapshot.toProto: every field maps to its own proto field" {
+    // Distinct values so a transposed pair (e.g. metric_bytes ↔ span_bytes)
+    // fails instead of silently reporting the wrong signal's volume.
+    const snap: VolumeSnapshot = .{
+        .log_records = 11,
+        .log_bytes = 22,
+        .metric_data_points = 33,
+        .metric_bytes = 44,
+        .spans = 55,
+        .span_bytes = 66,
+    };
+
+    const out = snap.toProto();
+    try testing.expectEqual(@as(i64, 11), out.log_records);
+    try testing.expectEqual(@as(i64, 22), out.log_bytes);
+    try testing.expectEqual(@as(i64, 33), out.metric_data_points);
+    try testing.expectEqual(@as(i64, 44), out.metric_bytes);
+    try testing.expectEqual(@as(i64, 55), out.spans);
+    try testing.expectEqual(@as(i64, 66), out.span_bytes);
+}
+
+test "VolumeSnapshot.isZero: any single tracked field defeats omission" {
+    try testing.expect((VolumeSnapshot{}).isZero());
+
+    // A consumer tracking only one signal, or only bytes, must still be
+    // reported — omission is reserved for "nothing observed at all".
+    inline for (@typeInfo(VolumeSnapshot).@"struct".fields) |f| {
+        var snap: VolumeSnapshot = .{};
+        @field(snap, f.name) = 1;
+        try testing.expect(!snap.isZero());
+    }
+}
+
+test "StatsCollector: volume seam is optional and no-ops when unwired" {
+    // A collector from a provider the registry never gave a volume fn to (or an
+    // older consumer building one by hand) must not crash and must report no
+    // volume.
+    var ctx: u8 = 0;
+    const collector: StatsCollector = .{
+        .context = &ctx,
+        .collect = struct {
+            fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot {
+                return &.{};
+            }
+        }.collect,
+    };
+
+    try testing.expect(collector.drainVolume().isZero());
+    try testing.expect(collector.drainVolume().isZero());
+}
diff --git a/src/policy/provider_http.zig b/src/policy/provider_http.zig
index c52366e..5b5bb1f 100644
--- a/src/policy/provider_http.zig
+++ b/src/policy/provider_http.zig
@@ -7,6 +7,7 @@ const o11y = @import("observability");
 const PolicyCallback = policy_provider.PolicyCallback;
 const StatsCollector = policy_provider.StatsCollector;
 const PolicyStatsSnapshot = policy_provider.PolicyStatsSnapshot;
+const VolumeSnapshot = policy_provider.VolumeSnapshot;
 const SyncRequest = proto.policy.SyncRequest;
 const SyncResponse = proto.policy.SyncResponse;
 const ClientMetadata = proto.policy.ClientMetadata;
@@ -35,10 +36,16 @@ const HttpPoliciesUnchanged = struct { reason: []const u8 };
 const HttpPolicyHashUpdated = struct { hash: []const u8 };
 const HttpPoliciesLoaded = struct { count: usize, url: []const u8, sync_timestamp: u64 };
 const HttpSyncRequestFailed = struct { url: []const u8, status: u16 };
+// A 200 whose SyncResponse carries `error_message`: the control plane rejected
+// the sync in-band, which is a failure even though the transport succeeded.
+const HttpSyncRejected = struct { url: []const u8, err: []const u8 };
 const HttpSyncRequestSucceeded = struct { url: []const u8, policy_statuses_sent: usize };
 // Emitted once per policy status about to be sent, so we can confirm at runtime
 // exactly which policies (and counts) are reported each sync.
 const HttpSyncStatusReported = struct { id: []const u8, match_hits: i64, match_misses: i64, errors: usize };
+// Emitted when a sync carries volume, for the same reason: confirming at
+// runtime what was reported. Omitted volume (all-zero) emits nothing.
+const HttpSyncVolumeReported = struct { volume: VolumeSnapshot };
 const HttpFetchStarted = struct {};
 const HttpFetchCompleted = struct {};
 
@@ -398,6 +405,12 @@ pub const HttpProvider = struct {
 
         const policy_statuses_list = try statsToSyncStatuses(temp_allocator, stats);
 
+        // Drain total observed volume (v1.7.1). Draining is the reset, exactly
+        // like the per-policy counters above: if this sync fails its interval is
+        // gone rather than replayed, since the server cannot tell a replay from
+        // new telemetry. Reported volume is a lower bound, not an exact total.
+        const volume: VolumeSnapshot = if (self.stats_collector) |c| c.drainVolume() else .{};
+
         // Log exactly what we're about to report, so runtime expectations can be
         // verified against the live snapshot.
         for (policy_statuses_list.items) |status| {
@@ -449,8 +462,16 @@ pub const HttpProvider = struct {
             .last_sync_timestamp_unix_nano = self.last_sync_timestamp,
             .last_successful_hash = last_hash,
             .policy_statuses = policy_statuses_list,
+            // Omitted rather than sent zero-valued when nothing was seen, per
+            // the spec's guidance for implementations that track no volume.
+            .volume = if (volume.isZero()) null else volume.toProto(),
         };
 
+        if (!volume.isZero()) {
+            const volume_event: HttpSyncVolumeReported = .{ .volume = volume };
+            self.bus.debug(volume_event);
+        }
+
         // Encode SyncRequest to JSON
         const request_body = try sync_request.jsonEncode(.{}, .{ .emit_oneof_field_name = false }, temp_allocator);
         // No defer needed - arena handles cleanup
@@ -526,6 +547,21 @@ pub const HttpProvider = struct {
             return err;
         };
 
+        // A 200 carrying `error_message` is an in-band rejection: the server did
+        // not accept this sync, so it must not advance the hash or deliver
+        // policies. Checked before touching sync state, as policy-go and
+        // policy-rs do.
+        if (parsed.value.error_message.len > 0) {
+            var mut = parsed;
+            defer mut.deinit();
+            const event: HttpSyncRejected = .{
+                .url = self.config_url,
+                .err = parsed.value.error_message,
+            };
+            self.bus.err(event);
+            return error.SyncRejected;
+        }
+
         return .{
             .parsed = parsed,
             .response_body = response_body,
@@ -649,6 +685,68 @@ test "HttpProvider: refreshClientClock advances stale now but leaves first reque
     try testing.expect(provider.http_client.now.?.nanoseconds > stale.nanoseconds);
 }
 
+/// Minimal control-plane stub: accept one POST, record its body, answer with
+/// `response` (200 and an empty SyncResponse by default). Serves exactly one
+/// request, so an unexpected extra sync is observable as a hang/failure.
+const SyncStub = struct {
+    allocator: std.mem.Allocator,
+    listener: std.Io.net.Server,
+    port: u16,
+    body: []u8 = &.{},
+    served: bool = false,
+    /// Response body. Set to something unparseable to exercise the decode
+    /// failure path.
+    response: []const u8 = "{}",
+
+    fn start(a: std.mem.Allocator, sio: std.Io) !SyncStub {
+        var attempt: u16 = 0;
+        while (attempt < 32) : (attempt += 1) {
+            const port: u16 = 42801 + attempt;
+            const addr = try std.Io.net.IpAddress.parse("127.0.0.1", port);
+            const listener = addr.listen(sio, .{}) catch |err| switch (err) {
+                error.AddressInUse => continue,
+                else => return err,
+            };
+            return .{ .allocator = a, .listener = listener, .port = port };
+        }
+        return error.AddressInUse;
+    }
+
+    fn deinit(self: *SyncStub, sio: std.Io) void {
+        defer self.* = undefined;
+        self.listener.deinit(sio);
+        if (self.body.len > 0) self.allocator.free(self.body);
+    }
+
+    fn serve(self: *SyncStub, sio: std.Io) void {
+        var stream = self.listener.accept(sio) catch return;
+        defer stream.close(sio);
+        var recv_buf: [16 * 1024]u8 = undefined;
+        var send_buf: [4 * 1024]u8 = undefined;
+        var conn_reader = stream.reader(sio, &recv_buf);
+        var conn_writer = stream.writer(sio, &send_buf);
+        var http_server = std.http.Server.init(&conn_reader.interface, &conn_writer.interface);
+        var request = http_server.receiveHead() catch return;
+        var transfer_buf: [8 * 1024]u8 = undefined;
+        const body_reader = request.readerExpectContinue(&transfer_buf) catch return;
+        self.body = body_reader.allocRemaining(self.allocator, .limited(1 << 20)) catch return;
+        self.served = true;
+        request.respond(self.response, .{ .status = .ok }) catch return;
+    }
+
+    /// Start a stub plus a provider pointed at it, in the shape every sync test
+    /// needs. Caller owns both (see the deinit pattern in the tests below).
+    fn provider(self: *const SyncStub, a: std.mem.Allocator, sio: std.Io, bus: *EventBus) !*HttpProvider {
+        var url_buf: [64]u8 = undefined;
+        const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/sync", .{self.port});
+        return HttpProvider.init(a, sio, bus, .{
+            .id = "p",
+            .url = url,
+            .poll_interval_seconds = 3600,
+        });
+    }
+};
+
 test "HttpProvider.close: flushes final stats once, then is idempotent" {
     const allocator = testing.allocator;
     // Own Threaded io so the stub control-plane server gets a real thread,
@@ -657,58 +755,9 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    // Minimal control-plane stub: accept one POST, record its body, answer 200
-    // with an empty SyncResponse. Refuses to serve more than one request so a
-    // second close() firing a redundant sync would be observable as a hang/fail.
-    const Stub = struct {
-        const Self = @This();
-
-        allocator: std.mem.Allocator,
-        listener: std.Io.net.Server,
-        port: u16,
-        body: []u8 = &.{},
-        served: bool = false,
-
-        fn start(a: std.mem.Allocator, sio: std.Io) !Self {
-            var attempt: u16 = 0;
-            while (attempt < 32) : (attempt += 1) {
-                const port: u16 = 42801 + attempt;
-                const addr = try std.Io.net.IpAddress.parse("127.0.0.1", port);
-                const listener = addr.listen(sio, .{}) catch |err| switch (err) {
-                    error.AddressInUse => continue,
-                    else => return err,
-                };
-                return .{ .allocator = a, .listener = listener, .port = port };
-            }
-            return error.AddressInUse;
-        }
-
-        fn deinit(self: *Self, sio: std.Io) void {
-            defer self.* = undefined;
-            self.listener.deinit(sio);
-            if (self.body.len > 0) self.allocator.free(self.body);
-        }
-
-        fn serve(self: *Self, sio: std.Io) void {
-            var stream = self.listener.accept(sio) catch return;
-            defer stream.close(sio);
-            var recv_buf: [16 * 1024]u8 = undefined;
-            var send_buf: [4 * 1024]u8 = undefined;
-            var conn_reader = stream.reader(sio, &recv_buf);
-            var conn_writer = stream.writer(sio, &send_buf);
-            var http_server = std.http.Server.init(&conn_reader.interface, &conn_writer.interface);
-            var request = http_server.receiveHead() catch return;
-            var transfer_buf: [8 * 1024]u8 = undefined;
-            const body_reader = request.readerExpectContinue(&transfer_buf) catch return;
-            self.body = body_reader.allocRemaining(self.allocator, .limited(1 << 20)) catch return;
-            self.served = true;
-            request.respond("{}", .{ .status = .ok }) catch return;
-        }
-    };
-
-    var stub = try Stub.start(allocator, io);
+    var stub = try SyncStub.start(allocator, io);
     defer stub.deinit(io);
-    var server_future = io.concurrent(Stub.serve, .{ &stub, io }) catch
+    var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch
         return error.SkipZigTest;
     defer server_future.cancel(io);
 
@@ -736,10 +785,17 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" {
             rows[0] = .{ .id = "hot", .hits = 5, .misses = 1 };
             return rows;
         }
+        fn drain(_: *anyopaque) VolumeSnapshot {
+            return .{ .log_records = 9, .log_bytes = 512 };
+        }
     };
     Collector.calls = 0;
     var ctx: u8 = 0;
-    provider.setStatsCollector(.{ .context = &ctx, .collect = Collector.collect });
+    provider.setStatsCollector(.{
+        .context = &ctx,
+        .collect = Collector.collect,
+        .collect_volume = Collector.drain,
+    });
 
     try provider.close();
     server_future.await(io);
@@ -750,6 +806,9 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" {
     // The final sync carried the tail stats: the policy id and its hit count.
     try testing.expect(std.mem.indexOf(u8, stub.body, "hot") != null);
     try testing.expect(std.mem.indexOf(u8, stub.body, "matchHits") != null);
+    // …and the observed volume.
+    try testing.expect(std.mem.indexOf(u8, stub.body, "\"logRecords\":\"9\"") != null);
+    try testing.expect(std.mem.indexOf(u8, stub.body, "\"logBytes\":\"512\"") != null);
 
     // Second close is a no-op: no extra collect, no extra request (the stub
     // only served one and its port is about to close).
@@ -785,6 +844,197 @@ test "HttpProvider.close: propagates the final sync error" {
     try provider.close();
 }
 
+test "HttpProvider: an all-zero volume is omitted from the sync request" {
+    const allocator = testing.allocator;
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var stub = try SyncStub.start(allocator, io);
+    defer stub.deinit(io);
+    var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch
+        return error.SkipZigTest;
+    defer server_future.cancel(io);
+
+    var noop_bus: o11y.NoopEventBus = undefined;
+    noop_bus.init(io);
+
+    var provider = try stub.provider(allocator, io, noop_bus.eventBus());
+    defer provider.deinit();
+
+    // Collector wired, but nothing observed yet: the spec says omit the message
+    // rather than send a zero-valued one.
+    var ctx: u8 = 0;
+    provider.setStatsCollector(.{
+        .context = &ctx,
+        .collect = struct {
+            fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot {
+                return &.{};
+            }
+        }.collect,
+        .collect_volume = struct {
+            fn drain(_: *anyopaque) VolumeSnapshot {
+                return .{};
+            }
+        }.drain,
+    });
+
+    try provider.close();
+    server_future.await(io);
+
+    try testing.expect(stub.served);
+    try testing.expect(std.mem.indexOf(u8, stub.body, "volume") == null);
+}
+
+test "HttpProvider: a provider with no volume seam wired syncs without volume" {
+    const allocator = testing.allocator;
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var stub = try SyncStub.start(allocator, io);
+    defer stub.deinit(io);
+    var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch
+        return error.SkipZigTest;
+    defer server_future.cancel(io);
+
+    var noop_bus: o11y.NoopEventBus = undefined;
+    noop_bus.init(io);
+
+    var provider = try stub.provider(allocator, io, noop_bus.eventBus());
+    defer provider.deinit();
+
+    // A collector built without the volume fns (an older consumer, or a
+    // provider the registry never handed them to) must still sync cleanly.
+    var ctx: u8 = 0;
+    provider.setStatsCollector(.{
+        .context = &ctx,
+        .collect = struct {
+            fn collect(arena: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot {
+                const rows = try arena.alloc(PolicyStatsSnapshot, 1);
+                rows[0] = .{ .id = "p", .hits = 1 };
+                return rows;
+            }
+        }.collect,
+    });
+
+    try provider.close();
+    server_future.await(io);
+
+    try testing.expect(stub.served);
+    // Per-policy stats still reported; volume simply absent.
+    try testing.expect(std.mem.indexOf(u8, stub.body, "matchHits") != null);
+    try testing.expect(std.mem.indexOf(u8, stub.body, "volume") == null);
+}
+
+test "HttpProvider: a 200 carrying error_message is a failed sync" {
+    const allocator = testing.allocator;
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var stub = try SyncStub.start(allocator, io);
+    defer stub.deinit(io);
+    // Well-formed SyncResponse carrying policies, a hash and a timestamp, but
+    // the control plane rejected the sync in-band. None of it may be adopted.
+    stub.response =
+        \\{"policies":[{"id":"p1","enabled":true}],
+        \\ "hash":"should-not-be-adopted",
+        \\ "syncTimestampUnixNano":"1234",
+        \\ "errorMessage":"client not provisioned"}
+    ;
+    var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch
+        return error.SkipZigTest;
+    defer server_future.cancel(io);
+
+    var noop_bus: o11y.NoopEventBus = undefined;
+    noop_bus.init(io);
+
+    var provider = try stub.provider(allocator, io, noop_bus.eventBus());
+    defer provider.deinit();
+
+    var ctx: u8 = 0;
+
+    // Wired directly rather than via subscribe(), which would fire its own
+    // initial fetch and consume the stub's single response.
+    const Sink = struct {
+        var notified: bool = false;
+        fn onUpdate(_: *anyopaque, _: policy_provider.PolicyUpdate) anyerror!void {
+            notified = true;
+            return;
+        }
+    };
+    Sink.notified = false;
+    provider.callback = .{ .context = &ctx, .onUpdate = Sink.onUpdate };
+
+    if (provider.close()) |_| {
+        try testing.expect(false); // an in-band rejection is not a success
+    } else |err| {
+        try testing.expectEqual(error.SyncRejected, err);
+    }
+    server_future.await(io);
+
+    // The rejected response's hash was not adopted as last-successful…
+    try testing.expect(provider.last_successful_hash == null);
+    // …its timestamp did not advance, so the next attempt is still a full sync…
+    try testing.expectEqual(@as(u64, 0), provider.last_sync_timestamp);
+    // …and its policies were never delivered to the registry.
+    try testing.expect(!Sink.notified);
+}
+
+test "HttpProvider: a failed sync drops its volume instead of replaying it" {
+    const allocator = testing.allocator;
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var noop_bus: o11y.NoopEventBus = undefined;
+    noop_bus.init(io);
+
+    // Stand-in for the registry's counters, draining on pull exactly as
+    // VolumeCounters.readAndReset does. There is deliberately no way to put a
+    // reading back: the server cannot tell a replay from new telemetry, so a
+    // failed sync's interval is lost rather than double counted.
+    const Counters = struct {
+        var pending: VolumeSnapshot = .{};
+        var drains: usize = 0;
+        fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot {
+            return &.{};
+        }
+        fn drain(_: *anyopaque) VolumeSnapshot {
+            drains += 1;
+            defer pending = .{};
+            return pending;
+        }
+    };
+    Counters.pending = .{ .log_records = 1, .log_bytes = 64, .spans = 1 };
+    Counters.drains = 0;
+
+    // Port 1 has nothing listening → the POST fails after the drain.
+    var provider = try HttpProvider.init(
+        allocator,
+        io,
+        noop_bus.eventBus(),
+        .{ .id = "p", .url = "http://127.0.0.1:1/sync", .poll_interval_seconds = 3600 },
+    );
+    defer provider.deinit();
+    var ctx: u8 = 0;
+    provider.setStatsCollector(.{
+        .context = &ctx,
+        .collect = Counters.collect,
+        .collect_volume = Counters.drain,
+    });
+
+    if (provider.close()) |_| {
+        try testing.expect(false); // expected the unreachable endpoint to error
+    } else |_| {}
+
+    // The interval was read into the failed request and is gone: reported volume
+    // is a lower bound on what was observed, never a replay.
+    try testing.expectEqual(@as(usize, 1), Counters.drains);
+    try testing.expect(Counters.pending.isZero());
+}
+
 test "statsToSyncStatuses: maps hits/misses/errors and reports every policy" {
     var arena = std.heap.ArenaAllocator.init(testing.allocator);
     defer arena.deinit();
diff --git a/src/policy/registry.zig b/src/policy/registry.zig
index e59d8be..2855d97 100644
--- a/src/policy/registry.zig
+++ b/src/policy/registry.zig
@@ -53,6 +53,67 @@ pub const PolicyAtomicStats = struct {
     }
 };
 
+// =============================================================================
+// Volume Counters (spec v1.7.1)
+// =============================================================================
+
+/// Total telemetry entering policy evaluation, regardless of match. Lives on
+/// the registry rather than the snapshot: a recompile allocates fresh
+/// per-policy stats, and volume must survive that to stay a valid denominator
+/// for hits/misses across a sync interval.
+///
+/// ponytail: one shared cache line of atomics, same shape as the per-policy
+/// hit/miss counters. If per-record contention ever shows up in a profile,
+/// shard per thread and sum on drain.
+pub const VolumeCounters = struct {
+    log_records: std.atomic.Value(i64) = .init(0),
+    log_bytes: std.atomic.Value(i64) = .init(0),
+    metric_data_points: std.atomic.Value(i64) = .init(0),
+    metric_bytes: std.atomic.Value(i64) = .init(0),
+    spans: std.atomic.Value(i64) = .init(0),
+    span_bytes: std.atomic.Value(i64) = .init(0),
+
+    /// Count one record of signal `T`. Called by the engine for every record
+    /// entering evaluation; consumers never call this directly.
+    pub inline fn record(self: *VolumeCounters, comptime T: policy_types.TelemetryType) void {
+        const counter = switch (T) {
+            .log => &self.log_records,
+            .metric => &self.metric_data_points,
+            .trace => &self.spans,
+        };
+        _ = counter.fetchAdd(1, .monotonic);
+    }
+
+    /// Add to the reported byte volume for signal `T`. Records are counted
+    /// automatically by `evaluate`; bytes are opt-in and must be the
+    /// uncompressed OTLP protobuf serialized size of the records as received
+    /// (an estimate is fine) — the engine reads records through accessors and
+    /// has no serialized form to measure. Leave it unreported rather than
+    /// reporting another encoding's size.
+    ///
+    /// Typically called once per received batch with the batch's serialized
+    /// size, not once per record.
+    pub inline fn addBytes(self: *VolumeCounters, comptime T: policy_types.TelemetryType, bytes: i64) void {
+        const counter = switch (T) {
+            .log => &self.log_bytes,
+            .metric => &self.metric_bytes,
+            .trace => &self.span_bytes,
+        };
+        _ = counter.fetchAdd(bytes, .monotonic);
+    }
+
+    /// Read and zero every counter, for reporting on a sync request. Reading is
+    /// the reset: a sync that then fails drops its interval rather than
+    /// replaying it, matching the per-policy counters above.
+    pub fn readAndReset(self: *VolumeCounters) policy_provider.VolumeSnapshot {
+        var out: policy_provider.VolumeSnapshot = .{};
+        inline for (@typeInfo(policy_provider.VolumeSnapshot).@"struct".fields) |f| {
+            @field(out, f.name) = @field(self, f.name).swap(0, .monotonic);
+        }
+        return out;
+    }
+};
+
 // =============================================================================
 // Observability Events
 // =============================================================================
@@ -258,6 +319,10 @@ pub const PolicyRegistry = struct {
     // subscribe for capability advertisement + broadcast-config routing.
     extension_sync_hooks: ?policy_provider.ExtensionSyncHooks,
 
+    // Total telemetry seen by the engine since the last successful sync
+    // (v1.7.1), independent of any policy or snapshot.
+    volume: VolumeCounters,
+
     /// Subscription context for provider callbacks.
     /// Allocated with stable address so the callback pointer remains valid.
     const Subscription = struct {
@@ -294,6 +359,7 @@ pub const PolicyRegistry = struct {
             .bus = bus,
             .extension_resolver = null,
             .extension_sync_hooks = null,
+            .volume = .{},
         };
     }
 
@@ -329,6 +395,7 @@ pub const PolicyRegistry = struct {
         prov.setStatsCollector(.{
             .context = self,
             .collect = collectStatsThunk,
+            .collect_volume = drainVolumeThunk,
         });
 
         if (self.extension_sync_hooks) |hooks| {
@@ -346,6 +413,11 @@ pub const PolicyRegistry = struct {
         return self.collectStats(arena);
     }
 
+    fn drainVolumeThunk(context: *anyopaque) policy_provider.VolumeSnapshot {
+        const self: *PolicyRegistry = @ptrCast(@alignCast(context));
+        return self.volume.readAndReset();
+    }
+
     /// Collect a stats row for every policy in the current snapshot, resetting
     /// the underlying atomic counters, and attach any compile errors recorded
     /// since the last recompile. Every policy is reported, including zero-hit
@@ -1744,6 +1816,85 @@ test "PolicyRegistry: collectStats drains hit/miss counters and reports every po
     try testing.expectEqual(@as(i64, 0), hot_row2.misses);
 }
 
+test "PolicyRegistry: volume counters survive recompiles and accept add-back" {
+    const allocator = testing.allocator;
+    var noop_bus: NoopEventBus = undefined;
+    noop_bus.init(std.Options.debug_io);
+    var registry = PolicyRegistry.init(allocator, noop_bus.eventBus());
+    defer registry.deinit();
+
+    registry.volume.record(.log);
+    registry.volume.addBytes(.log, 100);
+    registry.volume.record(.metric); // metric bytes left untracked
+    registry.volume.record(.trace);
+    registry.volume.addBytes(.trace, 40);
+
+    // A recompile allocates fresh per-policy stats; volume lives outside the
+    // snapshot and must not be zeroed by it.
+    var p = try createTestPolicy(allocator, "p");
+    defer freeTestPolicy(allocator, &p);
+    try registry.updatePolicies(&.{p}, "http-provider", .http);
+
+    const drained = registry.volume.readAndReset();
+    try testing.expectEqual(@as(i64, 1), drained.log_records);
+    try testing.expectEqual(@as(i64, 100), drained.log_bytes);
+    try testing.expectEqual(@as(i64, 1), drained.metric_data_points);
+    try testing.expectEqual(@as(i64, 0), drained.metric_bytes);
+    try testing.expectEqual(@as(i64, 1), drained.spans);
+    try testing.expectEqual(@as(i64, 40), drained.span_bytes);
+    try testing.expect(registry.volume.readAndReset().isZero());
+
+    // Reading is the reset: a drained interval is reported at most once, so
+    // only telemetry seen since the drain shows up next time.
+    registry.volume.record(.log);
+    registry.volume.addBytes(.log, 7);
+    const next = registry.volume.readAndReset();
+    try testing.expectEqual(@as(i64, 1), next.log_records);
+    try testing.expectEqual(@as(i64, 7), next.log_bytes);
+    try testing.expectEqual(@as(i64, 0), next.spans);
+}
+
+test "PolicyRegistry: subscribe wires the volume seam to this registry" {
+    const provider_http = @import("./provider_http.zig");
+
+    const allocator = testing.allocator;
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var noop_bus: NoopEventBus = undefined;
+    noop_bus.init(io);
+    var registry = PolicyRegistry.init(allocator, noop_bus.eventBus());
+    defer registry.deinit();
+
+    // Port 1 has nothing listening: subscribe's initial fetch fails and is
+    // warned about, which is fine — the wiring under test happens before it.
+    var provider = try provider_http.HttpProvider.init(
+        allocator,
+        io,
+        noop_bus.eventBus(),
+        .{ .id = "p", .url = "http://127.0.0.1:1/sync", .poll_interval_seconds = 3600 },
+    );
+    defer provider.deinit();
+    try registry.subscribe(.{ .http = provider });
+    // Stop the poll thread before the registry it drains goes away.
+    provider.shutdown();
+
+    // Nothing else in the suite proves subscribe passed the volume fn (the
+    // provider tests install collectors by hand), so assert the seam exists and
+    // that draining through it lands on *this* registry's counters.
+    const collector = provider.stats_collector orelse return error.NoCollector;
+    try testing.expect(collector.collect_volume != null);
+
+    registry.volume.record(.log);
+    registry.volume.addBytes(.log, 64);
+    const drained = collector.drainVolume();
+    try testing.expectEqual(@as(i64, 1), drained.log_records);
+    try testing.expectEqual(@as(i64, 64), drained.log_bytes);
+    // Draining through the seam reset the registry's counters.
+    try testing.expect(registry.volume.readAndReset().isZero());
+}
+
 test "PolicyRegistry: collectStats reports exactly the current policy set across updates" {
     const allocator = testing.allocator;
     var noop_bus: NoopEventBus = undefined;
diff --git a/src/proto/tero/policy/v1.pb.zig b/src/proto/tero/policy/v1.pb.zig
index 981994e..fe2ae01 100644
--- a/src/proto/tero/policy/v1.pb.zig
+++ b/src/proto/tero/policy/v1.pb.zig
@@ -2501,6 +2501,113 @@ pub const PolicySyncStatus = struct {
     }
 };
 
+/// VolumeStats reports the total telemetry a client observed since the last
+/// sync, regardless of whether any policy matched it. Counts are of records
+/// entering policy evaluation, before any keep or transform stage runs.
+///
+/// Counters are reset when they are read into a sync request, whether or not
+/// that sync then succeeds — the same rule PolicySyncStatus.match_hits and
+/// match_misses follow. A failed sync loses its interval from the numerator and
+/// the denominator alike, so match rates stay meaningful; counters from a failed
+/// sync must never be replayed, since the server cannot tell a replay from new
+/// telemetry. Reported volume is a lower bound, not an exact total.
+///
+/// Reporting volume is optional, and every field is individually optional: an
+/// implementation may report record counts without byte counts, or a subset of
+/// signals. Any field left at 0 means "not tracked" as much as it means "none
+/// seen", so consumers must not read 0 as an observation.
+///
+/// Byte counts, when reported, are the uncompressed OTLP protobuf serialized
+/// size of the records as received, and are an estimate; implementations that
+/// cannot measure this cheaply may approximate it. A size in any other encoding
+/// must not be reported here — leave the field at 0 instead.
+pub const VolumeStats = struct {
+    log_records: i64 = 0,
+    log_bytes: i64 = 0,
+    metric_data_points: i64 = 0,
+    metric_bytes: i64 = 0,
+    spans: i64 = 0,
+    span_bytes: i64 = 0,
+
+    pub const _desc_table = .{
+        .log_records = fd(1, .{ .scalar = .int64 }),
+        .log_bytes = fd(2, .{ .scalar = .int64 }),
+        .metric_data_points = fd(3, .{ .scalar = .int64 }),
+        .metric_bytes = fd(4, .{ .scalar = .int64 }),
+        .spans = fd(5, .{ .scalar = .int64 }),
+        .span_bytes = fd(6, .{ .scalar = .int64 }),
+    };
+
+    /// Encodes the message to the writer
+    /// The allocator is used to generate submessages internally.
+    /// Hence, an ArenaAllocator is a preferred choice if allocations are a bottleneck.
+    pub fn encode(
+        self: @This(),
+        writer: *std.Io.Writer,
+        allocator: std.mem.Allocator,
+    ) (std.Io.Writer.Error || std.mem.Allocator.Error)!void {
+        return protobuf.encode(writer, allocator, self);
+    }
+
+    /// Decodes the message from the bytes read from the reader.
+    pub fn decode(
+        reader: *std.Io.Reader,
+        allocator: std.mem.Allocator,
+    ) (protobuf.DecodingError || std.Io.Reader.Error || std.mem.Allocator.Error)!@This() {
+        return protobuf.decode(@This(), reader, allocator);
+    }
+
+    /// Deinitializes and frees the memory associated with the message.
+    pub fn deinit(self: *@This(), allocator: std.mem.Allocator) void {
+        return protobuf.deinit(allocator, self);
+    }
+
+    /// Duplicates the message.
+    pub fn dupe(self: @This(), allocator: std.mem.Allocator) std.mem.Allocator.Error!@This() {
+        return protobuf.dupe(@This(), self, allocator);
+    }
+
+    /// Decodes the message from the JSON string.
+    pub fn jsonDecode(
+        input: []const u8,
+        options: std.json.ParseOptions,
+        allocator: std.mem.Allocator,
+    ) !std.json.Parsed(@This()) {
+        return protobuf.json.decode(@This(), input, options, allocator);
+    }
+
+    /// Decodes the message from the JSON string, honoring pb options
+    /// (e.g. hex_bytes_fields for OTLP trace_id/span_id).
+    pub fn jsonDecodeOpts(
+        input: []const u8,
+        options: std.json.ParseOptions,
+        pb_options: protobuf.json.Options,
+        allocator: std.mem.Allocator,
+    ) !std.json.Parsed(@This()) {
+        return protobuf.json.decodeOpts(@This(), input, options, pb_options, allocator);
+    }
+
+    /// Encodes the message to a JSON string.
+    pub fn jsonEncode(
+        self: @This(),
+        options: std.json.Stringify.Options,
+        pb_options: protobuf.json.Options,
+        allocator: std.mem.Allocator,
+    ) ![]const u8 {
+        return protobuf.json.encode(self, options, pb_options, allocator);
+    }
+
+    /// This method is used by std.json
+    /// internally for deserialization. DO NOT RENAME!
+    pub fn jsonParse(
+        allocator: std.mem.Allocator,
+        source: anytype,
+        options: std.json.ParseOptions,
+    ) !@This() {
+        return protobuf.json.parse(@This(), allocator, source, options);
+    }
+};
+
 /// SyncRequest is sent by clients to request policy updates.
 pub const SyncRequest = struct {
     client_metadata: ?ClientMetadata = null,
@@ -2508,6 +2615,7 @@ pub const SyncRequest = struct {
     last_sync_timestamp_unix_nano: u64 = 0,
     last_successful_hash: []const u8 = &.{},
     policy_statuses: std.ArrayList(PolicySyncStatus) = .empty,
+    volume: ?VolumeStats = null,
 
     pub const _desc_table = .{
         .client_metadata = fd(1, .submessage),
@@ -2515,6 +2623,7 @@ pub const SyncRequest = struct {
         .last_sync_timestamp_unix_nano = fd(3, .{ .scalar = .fixed64 }),
         .last_successful_hash = fd(4, .{ .scalar = .string }),
         .policy_statuses = fd(5, .{ .repeated = .submessage }),
+        .volume = fd(6, .submessage),
     };
 
     /// Encodes the message to the writer