feat: update to policy v1.7.1

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

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 {

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;
 

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());
+}