From 62bc0565f5037472e395c3d60b58be70bbabaa74 Mon Sep 17 00:00:00 2001
From: jaronoff97 <jacobaronoff45@gmail.com>
Date: Fri, 10 Jul 2026 12:16:16 -0400
Subject: [PATCH 1/4] implement s3 dump extension

---
 Taskfile.yml                                |  40 ++++
 build.zig                                   |  26 +++
 build.zig.zon                               |   4 +-
 src/bench/datadog_log_bench.zig             |   1 +
 src/config/types.zig                        |  33 ++++
 src/frontend/exec.zig                       |   4 +
 src/root.zig                                |   1 +
 src/runtime/app.zig                         |  32 ++++
 src/runtime/extensions.zig                  | 191 ++++++++++++++++++++
 src/runtime/runtime_metrics.zig             |  67 ++++++-
 src/signals/datadog/logs.zig                |  27 ++-
 src/signals/datadog/metrics.zig             |   8 +-
 src/signals/otlp/logs.zig                   |   1 -
 src/signals/otlp/metrics.zig                |  10 +-
 src/signals/otlp/traces.zig                 |   1 -
 src/signals/prometheus/field_accessor.zig   |  10 +-
 src/signals/prometheus/streaming_filter.zig |   2 +-
 src/tail/eval_context.zig                   |   8 +-
 18 files changed, 448 insertions(+), 18 deletions(-)
 create mode 100644 src/runtime/extensions.zig

diff --git a/Taskfile.yml b/Taskfile.yml
index 3c4fd3b3..38949f15 100644
--- a/Taskfile.yml
+++ b/Taskfile.yml
@@ -298,12 +298,52 @@ tasks:
       - task: test
       - echo "✅ All checks passed"
 
+  test:s3-e2e:
+    desc: "s3-dump MinIO smoke test: dump a Datadog log to a real S3 server (requires docker)"
+    cmds:
+      - |
+        if ! command -v docker >/dev/null 2>&1; then
+          echo "docker is required for task test:s3-e2e"
+          exit 1
+        fi
+        CONTAINER=edge-minio-e2e
+        NETWORK=edge-minio-e2e-net
+        BUCKET=edge-e2e
+        cleanup() {
+          docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+          docker network rm "$NETWORK" >/dev/null 2>&1 || true
+        }
+        trap cleanup EXIT
+        cleanup
+        docker network create "$NETWORK" >/dev/null
+        docker run -d --rm --name "$CONTAINER" --network "$NETWORK" \
+          -p 9000:9000 \
+          -e MINIO_ROOT_USER=minioadmin \
+          -e MINIO_ROOT_PASSWORD=minioadmin \
+          minio/minio server /data >/dev/null
+        # Wait for the server to report healthy on the published port.
+        for i in $(seq 1 30); do
+          if curl -fsS http://127.0.0.1:9000/minio/health/live >/dev/null 2>&1; then break; fi
+          if [ "$i" = "30" ]; then echo "MinIO did not become healthy"; docker logs "$CONTAINER" || true; exit 1; fi
+          sleep 1
+        done
+        # Edge has no S3 admin client, so create the bucket with mc on the same
+        # docker network (reaches the server by container name).
+        docker run --rm --network "$NETWORK" --entrypoint /bin/sh minio/mc -c \
+          "mc alias set local http://$CONTAINER:9000 minioadmin minioadmin >/dev/null && mc mb -p local/$BUCKET >/dev/null"
+        AWS_ACCESS_KEY_ID=minioadmin \
+        AWS_SECRET_ACCESS_KEY=minioadmin \
+        S3_ENDPOINT=http://127.0.0.1:9000 \
+        S3_BUCKET="$BUCKET" \
+          zig build test-s3-e2e --summary all
+
   signoff:
     desc: "Sign off the codebase"
     aliases: ["sign", "s"]
     cmds:
       - task: do
       - task: build:safe
+      - task: test:s3-e2e
       - gh signoff
 
   install:
diff --git a/build.zig b/build.zig
index c3d35e2e..a2796ef2 100644
--- a/build.zig
+++ b/build.zig
@@ -61,6 +61,9 @@ pub fn build(b: *std.Build) void {
     // Shared modules from policy-zig ensure type identity across boundaries.
     const proto_mod = policy_dep.module("proto");
     const o11y_mod = policy_dep.module("observability");
+    // Optional extensions module (pulls in the z3 S3 client). Only edge code
+    // that wires s3-dump imports it.
+    const ext_mod = policy_dep.module("extensions");
 
     // ==========================================================================
     // Edge Library Module
@@ -74,6 +77,7 @@ pub fn build(b: *std.Build) void {
             .{ .name = "zimdjson", .module = zimdjson.module("zimdjson") },
             .{ .name = "policy_zig", .module = policy_dep.module("policy_zig") },
             .{ .name = "o11y", .module = o11y_mod },
+            .{ .name = "extensions", .module = ext_mod },
             .{ .name = "metrics_zig", .module = metrics_dep.module("metrics") },
             .{ .name = "httpz", .module = httpz_mod },
         },
@@ -111,6 +115,7 @@ pub fn build(b: *std.Build) void {
     exe.root_module.addImport("zimdjson", zimdjson.module("zimdjson"));
     exe.root_module.addImport("policy_zig", policy_dep.module("policy_zig"));
     exe.root_module.addImport("o11y", o11y_mod);
+    exe.root_module.addImport("extensions", ext_mod);
     exe.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
     exe.root_module.addImport("httpz", httpz_mod);
     exe.root_module.addOptions("build_options", build_options);
@@ -152,6 +157,7 @@ pub fn build(b: *std.Build) void {
         dist_exe.root_module.addImport("zimdjson", zimdjson.module("zimdjson"));
         dist_exe.root_module.addImport("policy_zig", policy_dep.module("policy_zig"));
         dist_exe.root_module.addImport("o11y", o11y_mod);
+        dist_exe.root_module.addImport("extensions", ext_mod);
         dist_exe.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
         dist_exe.root_module.addImport("httpz", httpz_mod);
         dist_exe.root_module.addOptions("build_options", build_options);
@@ -208,6 +214,26 @@ pub fn build(b: *std.Build) void {
     const test_step = b.step("test", "Run tests");
     test_step.dependOn(&run_mod_tests.step);
 
+    // Real-storage smoke test for the s3-dump extension, filtered to the MinIO
+    // e2e test. Excluded from `test` (it needs a live backend); driven by
+    // `task test:s3-e2e`, which starts MinIO, creates the bucket, and sets the
+    // S3 env vars this test reads.
+    const s3_e2e_tests = b.addTest(.{
+        .root_module = mod,
+        .filters = &.{"e2e minio"},
+    });
+    s3_e2e_tests.root_module.link_libc = true;
+    s3_e2e_tests.root_module.linkSystemLibrary("z", .{});
+    s3_e2e_tests.root_module.linkSystemLibrary("zstd", .{});
+    s3_e2e_tests.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
+    s3_e2e_tests.root_module.addOptions("build_options", build_options);
+    s3_e2e_tests.root_module.addAnonymousImport("otlp_metrics_benchmark_pb", .{
+        .root_source_file = b.path("bench/scaling/payloads/otlp-metrics.pb"),
+    });
+    const run_s3_e2e_tests = b.addRunArtifact(s3_e2e_tests);
+    const s3_e2e_step = b.step("test-s3-e2e", "Run the s3-dump MinIO smoke test (needs S3 env vars)");
+    s3_e2e_step.dependOn(&run_s3_e2e_tests.step);
+
     // ==========================================================================
     // Benchmark Tools
     // ==========================================================================
diff --git a/build.zig.zon b/build.zig.zon
index 5a62cfc3..192116fc 100644
--- a/build.zig.zon
+++ b/build.zig.zon
@@ -49,8 +49,8 @@
             .hash = "zbench-0.13.0-YTdc7xVBAQCCMC-IdLLuotBeiNNNm8k9Pi2V4VYqrwfI",
         },
         .policy_zig = .{
-            .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.4.4.tar.gz",
-            .hash = "policy_zig-0.1.7-5_dp3q0-EgDyYRR7_LInB9k-CjcBiI8nGewWKOXAVbvO",
+            .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.5.0.tar.gz",
+            .hash = "policy_zig-0.1.7-5_dp3j7lFAAWNWhq8DkCv990ujiTWKWjYxzsIKNpeXIT",
         },
     },
     .paths = .{
diff --git a/src/bench/datadog_log_bench.zig b/src/bench/datadog_log_bench.zig
index c0846579..b30bb0fa 100644
--- a/src/bench/datadog_log_bench.zig
+++ b/src/bench/datadog_log_bench.zig
@@ -267,6 +267,7 @@ fn evalOne(registry: *PolicyRegistry, record: []const u8) std.meta.Tag(logs.Reco
         registry,
         g_bus,
         record,
+        null,
     ) catch |err| std.debug.panic("evalLogRecord failed: {t}", .{err});
     return std.meta.activeTag(verdict);
 }
diff --git a/src/config/types.zig b/src/config/types.zig
index cb6ee478..3815250d 100644
--- a/src/config/types.zig
+++ b/src/config/types.zig
@@ -24,6 +24,36 @@ pub const PrometheusModuleConfig = struct {
     max_output_bytes_per_scrape: usize = 10 * 1024 * 1024,
 };
 
+/// A single S3-compatible destination the s3-dump extension can write to.
+/// Fields (except `name`) mirror the z3 target config the extension expects;
+/// `name` is what policies reference via `ExtensionTargetRef`.
+pub const S3TargetConfig = struct {
+    name: []const u8,
+    /// S3-compatible endpoint URL. Null uses the AWS default for `region`.
+    endpoint: ?[]const u8 = null,
+    region: []const u8 = "us-east-1",
+    bucket: []const u8,
+    prefix: []const u8 = "",
+    /// Path-style (MinIO/R2) vs virtual-host-style URLs.
+    force_path_style: bool = true,
+};
+
+/// `com.usetero/s3-dump` extension configuration. Off unless `enabled` and at
+/// least one target is set. Credentials are NOT here — they come from the
+/// AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY environment variables.
+pub const S3DumpConfig = struct {
+    enabled: bool = false,
+    /// Interval between background flush cycles (single flush task).
+    flush_interval_ms: u64 = 30_000,
+    // Batching knobs — map 1:1 onto the extension's S3Dump.Options.
+    max_batch_bytes: usize = 4 << 20,
+    max_batch_records: u32 = 10_000,
+    max_batch_age_ms: u64 = 30_000,
+    max_sealed_bytes: usize = 32 << 20,
+    max_attempts: usize = 1,
+    targets: []const S3TargetConfig = &.{},
+};
+
 /// Main proxy configuration - loadable via zonfig
 pub const ProxyConfig = struct {
     // Network config
@@ -69,4 +99,7 @@ pub const ProxyConfig = struct {
 
     // Module-specific configuration
     prometheus: PrometheusModuleConfig = .{},
+
+    /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default.
+    s3_dump: S3DumpConfig = .{},
 };
diff --git a/src/frontend/exec.zig b/src/frontend/exec.zig
index f10c7a39..02f2de21 100644
--- a/src/frontend/exec.zig
+++ b/src/frontend/exec.zig
@@ -68,6 +68,9 @@ pub const SharedCtx = struct {
     limits: limits_mod.Limits,
     /// Debug tap, or null when disabled by config. See `TapState`.
     tap: ?*TapState = null,
+    /// Extension dispatch sink (s3-dump), or null when extensions are off.
+    /// Threaded into per-record policy evaluation on the Datadog log path.
+    extension_sink: ?policy.ExtensionSink = null,
 };
 
 /// Routes and plans a request from transport-neutral parts. Returns null
@@ -411,6 +414,7 @@ pub const RecordSink = struct {
             self.ctx.registry,
             self.ctx.bus,
             bytes,
+            self.ctx.extension_sink,
         );
         switch (verdict) {
             .keep => return .keep,
diff --git a/src/root.zig b/src/root.zig
index 1815e367..491e809e 100644
--- a/src/root.zig
+++ b/src/root.zig
@@ -100,6 +100,7 @@ test {
     _ = @import("frontend/stdio/server.zig");
     _ = @import("frontend/httpz/server.zig");
     _ = @import("runtime/distro.zig");
+    _ = @import("runtime/extensions.zig");
     _ = @import("signals/prometheus/root.zig");
     _ = @import("lambda/root.zig");
     _ = @import("zonfig/root.zig");
diff --git a/src/runtime/app.zig b/src/runtime/app.zig
index c6afa069..3947754b 100644
--- a/src/runtime/app.zig
+++ b/src/runtime/app.zig
@@ -22,6 +22,7 @@ const router_mod = @import("../service/router.zig");
 const upstream_mod = @import("../frontend/upstream.zig");
 const exec_mod = @import("../frontend/exec.zig");
 const frontend_select = @import("../frontend/select.zig");
+const ext_rt = @import("extensions.zig");
 
 const policy = @import("policy_zig");
 const o11y = @import("o11y");
@@ -223,6 +224,7 @@ pub const EngineOptions = struct {
     metrics_url: ?[]const u8 = null,
     service_options: distro.ServiceOptions = .{},
     tap_enabled: bool = false,
+    extension_sink: ?policy.ExtensionSink = null,
 };
 
 pub const Engine = struct {
@@ -310,6 +312,7 @@ pub const Engine = struct {
             .metrics = metrics,
             .limits = self.limits,
             .tap = if (options.tap_enabled) &self.tap else null,
+            .extension_sink = options.extension_sink,
         };
 
         self.server = try .init(
@@ -449,6 +452,18 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
     var registry = policy.Registry.init(allocator, bus);
     defer registry.deinit();
 
+    // s3-dump extension: wired before the loader subscribes providers so the
+    // sync hooks reach every provider. Stays inert unless enabled with targets.
+    var exts = ext_rt.Extensions.init(allocator, .{ .log = ext_rt.datadogLogEncode });
+    defer exts.deinit();
+    const s3_dump_active = config.s3_dump.enabled and config.s3_dump.targets.len > 0;
+    var extension_sink: ?policy.ExtensionSink = null;
+    if (s3_dump_active) {
+        try ext_rt.configure(&exts, allocator, io, config.s3_dump, init.environ_map);
+        exts.register(&registry);
+        extension_sink = exts.sink();
+    }
+
     var runtime_metrics = try RuntimeMetrics.init(allocator, io, distributionLabel(distribution));
     defer runtime_metrics.deinit();
     runtime_metrics.setBuildInfo(build_options.version, build_options.commit);
@@ -475,6 +490,7 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
             .prometheus_max_output_bytes = config.prometheus.max_output_bytes_per_scrape,
         },
         .tap_enabled = config.tap_enabled,
+        .extension_sink = extension_sink,
     });
     defer engine.destroy();
 
@@ -490,6 +506,17 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
 
     try engine.start();
 
+    // Single background flush task, torn down with the rest of the group on
+    // shutdown; the final force-flush runs on the main path below.
+    if (s3_dump_active) {
+        try engine.lifecycle.spawn(io, ext_rt.flushLoop, .{
+            &exts,
+            io,
+            config.s3_dump.flush_interval_ms,
+            &runtime_metrics,
+        });
+    }
+
     // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
     bus.info(DataPlaneBudget{
         .max_connections = engine.limits.max_connections,
@@ -506,6 +533,11 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
     engine.awaitShutdown();
     engine.stop();
 
+    // Drain any buffered dump batches before we tear down (io is still live).
+    if (s3_dump_active) {
+        runtime_metrics.recordS3DumpFlush(exts.flush(io, .{ .force = true }));
+    }
+
     if (signal_waiter) |waiter| {
         shutdown_waiter.store(true, .release);
         std.posix.kill(std.c.getpid(), std.posix.SIG.USR1) catch |err|
diff --git a/src/runtime/extensions.zig b/src/runtime/extensions.zig
new file mode 100644
index 00000000..a044bfb9
--- /dev/null
+++ b/src/runtime/extensions.zig
@@ -0,0 +1,191 @@
+//! Wiring for the `com.usetero/s3-dump` policy extension.
+//!
+//! Today this dumps Datadog JSON logs to S3-compatible storage. The engine
+//! classifies each record per policy binding (kept/dropped/unmatched) and
+//! hands the selected records to the extension sink, which serializes them
+//! into batches; a single background task flushes sealed batches to S3.
+//!
+//! Only the Datadog log path is wired: `Extensions` carries one encoder per
+//! signal, and within a binary a signal maps to exactly one accessor context
+//! type. Datadog logs use `dd_logs.FieldAccessorContext`, so that is the type
+//! `datadogLogEncode` casts back to. Other signal sources leave the sink null
+//! (fail-open, zero cost).
+
+const std = @import("std");
+const ext = @import("extensions");
+const proto = @import("proto");
+const dd_logs = @import("../signals/datadog/logs.zig");
+const datadog_log = @import("../signals/datadog/log.zig");
+const config = @import("../config/types.zig");
+const runtime_metrics = @import("runtime_metrics.zig");
+
+pub const Extensions = ext.Extensions;
+
+// Routed into the EventBus via StdLogAdapter (see app.zig).
+const log = std.log.scoped(.s3_dump);
+
+/// Encoder for `.log` records on the Datadog path. The engine passes the
+/// accessor context (`*FieldAccessorContext`) borrowed for the duration of
+/// this call; we render its log as one JSON object (the handler appends the
+/// newline that makes the batch ndjson).
+pub fn datadogLogEncode(record: *const anyopaque, writer: *std.Io.Writer) anyerror!void {
+    const field_ctx: *const dd_logs.FieldAccessorContext = @ptrCast(@alignCast(record));
+    try std.json.Stringify.value(field_ctx.log.*, .{}, writer);
+}
+
+/// Read S3 credentials from the environment. Returns null when either half is
+/// absent — the extension then counts-and-drops deliveries for its targets
+/// (fail-open), so a missing credential never stalls telemetry.
+pub fn credentialsFromEnv(env: *const std.process.Environ.Map) ?ext.S3Dump.Credentials {
+    const access = env.get("AWS_ACCESS_KEY_ID") orelse return null;
+    const secret = env.get("AWS_SECRET_ACCESS_KEY") orelse return null;
+    if (access.len == 0 or secret.len == 0) return null;
+    return .{ .access_key_id = access, .secret_access_key = secret };
+}
+
+/// Enable the s3-dump handler on `exts` from config, wiring credentials and
+/// every configured target. Caller must have `init`ed `exts` with encoders and
+/// must `register` it with the registry afterwards. Targets that fail to parse
+/// are logged and skipped (fail-open); a target set that ends up empty leaves
+/// the handler enabled but inert.
+pub fn configure(
+    exts: *Extensions,
+    gpa: std.mem.Allocator,
+    io: std.Io,
+    cfg: config.S3DumpConfig,
+    env: *const std.process.Environ.Map,
+) !void {
+    const creds = credentialsFromEnv(env);
+    if (creds == null) log.warn("AWS credentials not set; deliveries will drop", .{});
+
+    const handler = exts.enableS3Dump(gpa, .{
+        .max_batch_bytes = cfg.max_batch_bytes,
+        .max_batch_records = cfg.max_batch_records,
+        .max_batch_age_ms = cfg.max_batch_age_ms,
+        .max_sealed_bytes = cfg.max_sealed_bytes,
+        .max_attempts = cfg.max_attempts,
+    }, creds);
+
+    for (cfg.targets) |target| {
+        // z3 target config = the target fields minus the policy-facing `name`.
+        var json: std.Io.Writer.Allocating = .init(gpa);
+        defer json.deinit();
+        try std.json.Stringify.value(.{
+            .endpoint = target.endpoint,
+            .region = target.region,
+            .bucket = target.bucket,
+            .prefix = target.prefix,
+            .force_path_style = target.force_path_style,
+        }, .{}, &json.writer);
+        handler.addTarget(io, target.name, json.written()) catch |err| {
+            log.warn("skipping target '{s}': {s}", .{ target.name, @errorName(err) });
+        };
+    }
+}
+
+/// Background flush task: seal-and-upload on an interval until the lifecycle
+/// group cancels the sleep (shutdown). The final force-flush is done by the
+/// caller on the main path, where `io` is still live.
+pub fn flushLoop(
+    exts: *Extensions,
+    io: std.Io,
+    interval_ms: u64,
+    metrics: ?*runtime_metrics.RuntimeMetrics,
+) void {
+    const interval: std.Io.Duration = .fromMilliseconds(@intCast(interval_ms));
+    while (true) {
+        std.Io.sleep(io, interval, .awake) catch return; // canceled at shutdown
+        const result = exts.flush(io, .{});
+        if (metrics) |mx| mx.recordS3DumpFlush(result);
+    }
+}
+
+test "datadogLogEncode renders the delivered log context as JSON" {
+    const allocator = std.testing.allocator;
+    var ddlog = try datadog_log.DatadogLog.parseRaw(allocator, "{\"status\":\"info\",\"message\":\"hi\"}");
+    defer ddlog.deinit(allocator);
+
+    // The engine hands the encoder the accessor context, borrowed for the call.
+    var ctx: dd_logs.FieldAccessorContext = .{ .log = &ddlog, .allocator = allocator };
+
+    var out: std.Io.Writer.Allocating = .init(allocator);
+    defer out.deinit();
+    try datadogLogEncode(&ctx, &out.writer);
+
+    // Round-trips through the same serializer as the forwarding path.
+    try std.testing.expect(std.mem.indexOf(u8, out.written(), "\"message\":\"hi\"") != null);
+}
+
+// Real-storage smoke test: drives edge's encoder + the s3-dump handler + a
+// real flush against a MinIO container, exactly the path a `mode: dropped`
+// policy takes. Self-skips unless the S3 env vars are set, so it's inert in
+// the normal suite; `task test:s3-e2e` starts MinIO, creates the bucket, sets
+// the env, and runs `zig build test-s3-e2e` (which filters to this test).
+//
+// Verification is by the flush result: MinIO validates the payload SHA-256 z3
+// sends, so `objects_uploaded == 1 && objects_failed == 0` means the bytes
+// edge produced were accepted intact. Byte-exact readback is covered by the
+// encoder test above and by policy-zig's own e2e.
+test "e2e minio: edge datadog log dump uploads to a real S3 server" {
+    const env = std.testing.environ;
+    const access = env.getPosix("AWS_ACCESS_KEY_ID") orelse return error.SkipZigTest;
+    const secret = env.getPosix("AWS_SECRET_ACCESS_KEY") orelse return error.SkipZigTest;
+    const endpoint = env.getPosix("S3_ENDPOINT") orelse "http://127.0.0.1:9000";
+    const bucket = env.getPosix("S3_BUCKET") orelse "edge-e2e";
+    const gpa = std.testing.allocator;
+
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var exts = Extensions.init(gpa, .{ .log = datadogLogEncode });
+    defer exts.deinit();
+    const handler = exts.enableS3Dump(gpa, .{ .max_attempts = 1 }, .{
+        .access_key_id = access,
+        .secret_access_key = secret,
+    });
+
+    const target_json = try std.fmt.allocPrint(
+        gpa,
+        \\{{"endpoint":"{s}","region":"us-east-1","bucket":"{s}","prefix":"e2e/","force_path_style":true}}
+    ,
+        .{ endpoint, bucket },
+    );
+    defer gpa.free(target_json);
+    try handler.addTarget(io, "minio", target_json);
+
+    // Resolve a delivery slot the way the engine does at snapshot compile:
+    // the extension config bytes are a serialized ExtensionTargetRef.
+    var ref_w: std.Io.Writer.Allocating = .init(gpa);
+    defer ref_w.deinit();
+    const ref: proto.policy.ExtensionTargetRef = .{ .kind = "s3", .name = "minio" };
+    try ref.encode(&ref_w.writer, gpa);
+    const slot = handler.resolve(io, .log, "e2e-policy", ref_w.written()) orelse
+        return error.TargetNotResolved;
+
+    // A real Datadog log, delivered through edge's own encoder.
+    var ddlog = try datadog_log.DatadogLog.parseRaw(gpa, "{\"status\":\"info\",\"message\":\"e2e\"}");
+    defer ddlog.deinit(gpa);
+    var ctx: dd_logs.FieldAccessorContext = .{ .log = &ddlog, .allocator = gpa };
+    handler.deliver(io, slot, &ctx, datadogLogEncode);
+
+    const result = exts.flush(io, .{ .force = true });
+    try std.testing.expectEqual(@as(u32, 1), result.objects_uploaded);
+    try std.testing.expectEqual(@as(u32, 0), result.objects_failed);
+    try std.testing.expectEqual(@as(u64, 1), result.records_uploaded);
+
+    // The flush stats must surface on the /_edge/metrics endpoint. Record the
+    // result the way the flush loop does and scrape the rendered output.
+    var metrics = try runtime_metrics.RuntimeMetrics.init(gpa, io, .edge);
+    defer metrics.deinit();
+    metrics.recordS3DumpFlush(result);
+
+    var scrape: std.Io.Writer.Allocating = .init(gpa);
+    defer scrape.deinit();
+    try metrics.writePrometheus(&scrape.writer);
+    const out = scrape.written();
+    try std.testing.expect(std.mem.indexOf(u8, out, "edge_s3_dump_flushes_total 1") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "edge_s3_dump_objects_uploaded_total 1") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "edge_s3_dump_objects_failed_total 0") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "edge_s3_dump_records_uploaded_total 1") != null);
+}
diff --git a/src/runtime/runtime_metrics.zig b/src/runtime/runtime_metrics.zig
index da8d1556..8b5f3d85 100644
--- a/src/runtime/runtime_metrics.zig
+++ b/src/runtime/runtime_metrics.zig
@@ -1,5 +1,6 @@
 const std = @import("std");
 const m = @import("metrics_zig");
+const ext = @import("extensions");
 
 const log = std.log.scoped(.runtime_metrics);
 
@@ -121,6 +122,16 @@ const InternalMetrics = struct {
     edge_policies_loaded: PoliciesLoaded,
     edge_build_info: BuildInfo,
 
+    // s3-dump extension flush stats (aggregate across targets; the FlushResult
+    // the handler returns is already summed, so these carry no labels).
+    edge_s3_dump_flushes_total: S3DumpFlushes,
+    edge_s3_dump_objects_uploaded_total: S3DumpCounter,
+    edge_s3_dump_objects_failed_total: S3DumpCounter,
+    edge_s3_dump_records_uploaded_total: S3DumpCounter,
+    edge_s3_dump_records_dropped_total: S3DumpCounter,
+    edge_s3_dump_bytes_uploaded_total: S3DumpCounter,
+    edge_s3_dump_backlog_bytes: S3DumpBacklog,
+
     const RequestsTotal = m.CounterVec(u64, RequestLabels);
     const RequestDurationSeconds = m.HistogramVec(
         f64,
@@ -135,6 +146,11 @@ const InternalMetrics = struct {
     const PolicyRecordsDroppedTotal = m.CounterVec(u64, PolicyLabels);
     const PoliciesLoaded = m.GaugeVec(u64, PolicySignalLabels);
     const BuildInfo = m.GaugeVec(u64, BuildInfoLabels);
+
+    // Label-less: one series each, so plain Counter/Gauge (no allocation).
+    const S3DumpFlushes = m.Counter(u64);
+    const S3DumpCounter = m.Counter(u64);
+    const S3DumpBacklog = m.Gauge(u64);
 };
 
 pub const RuntimeMetrics = struct {
@@ -217,6 +233,41 @@ pub const RuntimeMetrics = struct {
                     .{ .help = "Build metadata for this edge process." },
                     .{},
                 ),
+                .edge_s3_dump_flushes_total = InternalMetrics.S3DumpFlushes.init(
+                    "edge_s3_dump_flushes_total",
+                    .{ .help = "Total number of s3-dump flush cycles executed." },
+                    .{},
+                ),
+                .edge_s3_dump_objects_uploaded_total = InternalMetrics.S3DumpCounter.init(
+                    "edge_s3_dump_objects_uploaded_total",
+                    .{ .help = "Total s3-dump objects successfully uploaded." },
+                    .{},
+                ),
+                .edge_s3_dump_objects_failed_total = InternalMetrics.S3DumpCounter.init(
+                    "edge_s3_dump_objects_failed_total",
+                    .{ .help = "Total s3-dump object uploads that failed." },
+                    .{},
+                ),
+                .edge_s3_dump_records_uploaded_total = InternalMetrics.S3DumpCounter.init(
+                    "edge_s3_dump_records_uploaded_total",
+                    .{ .help = "Total telemetry records uploaded via s3-dump." },
+                    .{},
+                ),
+                .edge_s3_dump_records_dropped_total = InternalMetrics.S3DumpCounter.init(
+                    "edge_s3_dump_records_dropped_total",
+                    .{ .help = "Total records dropped by s3-dump (backlog full, encode failure, or no credentials)." },
+                    .{},
+                ),
+                .edge_s3_dump_bytes_uploaded_total = InternalMetrics.S3DumpCounter.init(
+                    "edge_s3_dump_bytes_uploaded_total",
+                    .{ .help = "Total object body bytes uploaded via s3-dump." },
+                    .{},
+                ),
+                .edge_s3_dump_backlog_bytes = InternalMetrics.S3DumpBacklog.init(
+                    "edge_s3_dump_backlog_bytes",
+                    .{ .help = "Sealed-but-not-yet-uploaded s3-dump bytes after the last flush." },
+                    .{},
+                ),
             },
         };
         try metrics.initializeStaticSeries();
@@ -273,7 +324,9 @@ pub const RuntimeMetrics = struct {
         // callers deinit RuntimeMetrics only after the server has stopped
         // (listen thread joined), so no worker threads touch the metrics here.
         inline for (std.meta.fields(InternalMetrics)) |field| {
-            @field(self.internal, field.name).deinit();
+            // Vec metrics allocate per-series and expose deinit; the label-less
+            // plain Counter/Gauge don't allocate and have none.
+            if (@hasDecl(field.type, "deinit")) @field(self.internal, field.name).deinit();
         }
         self.* = undefined;
     }
@@ -282,6 +335,18 @@ pub const RuntimeMetrics = struct {
         try m.write(&self.internal, writer);
     }
 
+    /// Fold one s3-dump flush result into the extension metrics. Called after
+    /// every flush (the background loop and the shutdown drain).
+    pub fn recordS3DumpFlush(self: *RuntimeMetrics, result: ext.S3Dump.FlushResult) void {
+        self.internal.edge_s3_dump_flushes_total.incr();
+        self.internal.edge_s3_dump_objects_uploaded_total.incrBy(result.objects_uploaded);
+        self.internal.edge_s3_dump_objects_failed_total.incrBy(result.objects_failed);
+        self.internal.edge_s3_dump_records_uploaded_total.incrBy(result.records_uploaded);
+        self.internal.edge_s3_dump_records_dropped_total.incrBy(result.records_dropped);
+        self.internal.edge_s3_dump_bytes_uploaded_total.incrBy(result.bytes_uploaded);
+        self.internal.edge_s3_dump_backlog_bytes.set(@intCast(result.backlog_bytes));
+    }
+
     pub fn recordRequest(
         self: *RuntimeMetrics,
         method: MethodLabel,
diff --git a/src/signals/datadog/logs.zig b/src/signals/datadog/logs.zig
index 80905adf..e5153696 100644
--- a/src/signals/datadog/logs.zig
+++ b/src/signals/datadog/logs.zig
@@ -129,6 +129,7 @@ pub fn evalLogRecord(
     registry: *const PolicyRegistry,
     bus: *EventBus,
     record: []const u8,
+    sink: ?policy.ExtensionSink,
 ) !RecordVerdict {
     // Single-pass zero-copy parse; zimdjson never runs on the hot path.
     // Anything the walker doesn't like — non-object records, escaped keys,
@@ -144,7 +145,7 @@ pub fn evalLogRecord(
 
     const engine = PolicyEngine.init(bus, @constCast(registry));
     var policy_id_buf: [MAX_MATCHES_PER_SCAN][]const u8 = undefined;
-    const result = filterLog(&engine, &log_obj, scratch, &policy_id_buf);
+    const result = filterLog(&engine, &log_obj, scratch, &policy_id_buf, sink);
     if (!result.keep) return .drop;
     if (!result.mutated) return .keep;
 
@@ -206,6 +207,13 @@ fn lookupLogAttribute(log: *DatadogLog, allocator: std.mem.Allocator, path: []co
 /// Datadog logs have fields at the root level: message, status/level, ddtags, service, etc.
 /// All attribute types (log, resource, scope) search the same flat namespace since
 /// Datadog has no OTLP-style resource/scope hierarchy.
+/// Typed read primitive (required since v0.5.0). Datadog log fields are all
+/// string-valued in the JSON, so wrap the byte primitive as `.string` —
+/// byte-identical to the pre-v0.5.0 `value` path.
+pub fn logTypedValue(ctx: *const anyopaque, field: FieldRef) ?policy.TypedValue {
+    return .{ .string = logValue(ctx, field) orelse return null };
+}
+
 pub fn logValue(ctx: *const anyopaque, field: FieldRef) ?[]const u8 {
     const field_ctx: *const FieldAccessorContext = @ptrCast(@alignCast(ctx));
     const log = field_ctx.log;
@@ -320,7 +328,7 @@ pub fn logDelete(ctx: *anyopaque, field: FieldRef) bool {
 /// support rename, so `move` is left null (policies that require rename are
 /// rejected at snapshot-compile time).
 pub const log_accessor: policy.LogAccessor = .{
-    .value = logValue,
+    .typed_value = logTypedValue,
     .set = logSet,
     .delete = logDelete,
 };
@@ -338,6 +346,7 @@ fn filterLog(
     log: *DatadogLog,
     allocator: std.mem.Allocator,
     policy_id_buf: [][]const u8,
+    sink: ?policy.ExtensionSink,
 ) FilterLogResult {
     var field_ctx: FieldAccessorContext = .{ .log = log, .allocator = allocator };
     const result = engine.evaluate(
@@ -345,7 +354,7 @@ fn filterLog(
         &log_accessor,
         &field_ctx,
         policy_id_buf,
-        .{ .scratch = allocator, .io = engine.bus.io },
+        .{ .scratch = allocator, .io = engine.bus.io, .extension_sink = sink },
     );
     // Re-serialize the wrapped message once if a transform edited inside it.
     log.finalizeWrapped(allocator);
@@ -472,7 +481,7 @@ fn processJsonLogsWithFilter(
                 };
 
                 state.original_count += 1;
-                const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf);
+                const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, null);
                 if (filter_result.mutated) state.mutated = true;
                 if (filter_result.keep) {
                     try state.kept.append(arena, log_obj);
@@ -489,7 +498,7 @@ fn processJsonLogsWithFilter(
             };
 
             state.original_count = 1;
-            const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf);
+            const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, null);
             if (filter_result.mutated) state.mutated = true;
             if (filter_result.keep) {
                 try state.kept.append(arena, log_obj);
@@ -809,6 +818,7 @@ test "evalLogRecord - no policies keeps record without output" {
         noop_bus.eventBus(),
         \\{"status": "info", "message": "test1"}
         ,
+        null,
     );
     try std.testing.expectEqual(RecordVerdict.keep, verdict);
 }
@@ -847,6 +857,7 @@ test "evalLogRecord - drop policy drops, non-matching keeps" {
         noop_bus.eventBus(),
         \\{"status": "debug", "message": "debug msg"}
         ,
+        null,
     );
     try std.testing.expectEqual(RecordVerdict.drop, dropped);
 
@@ -859,6 +870,7 @@ test "evalLogRecord - drop policy drops, non-matching keeps" {
         noop_bus.eventBus(),
         \\{"status": "error", "message": "error msg"}
         ,
+        null,
     );
     try std.testing.expectEqual(RecordVerdict.keep, kept);
 }
@@ -906,6 +918,7 @@ test "evalLogRecord - transform yields replace with serialized record" {
         noop_bus.eventBus(),
         \\{"message": "test log message", "service": "my-service", "status": "info"}
         ,
+        null,
     );
     try std.testing.expect(verdict == .replace);
     try std.testing.expect(std.mem.indexOf(u8, verdict.replace, "test log message") != null);
@@ -927,10 +940,10 @@ test "evalLogRecord - malformed and non-object records fail open to keep" {
     defer arena.deinit();
 
     const bus = noop_bus.eventBus();
-    const malformed = try evalLogRecord(arena.allocator(), &parser, allocator, &registry, bus, "{ not json }");
+    const malformed = try evalLogRecord(arena.allocator(), &parser, allocator, &registry, bus, "{ not json }", null);
     try std.testing.expectEqual(RecordVerdict.keep, malformed);
 
-    const scalar = try evalLogRecord(arena.allocator(), &parser, allocator, &registry, bus, "42");
+    const scalar = try evalLogRecord(arena.allocator(), &parser, allocator, &registry, bus, "42", null);
     try std.testing.expectEqual(RecordVerdict.keep, scalar);
 }
 
diff --git a/src/signals/datadog/metrics.zig b/src/signals/datadog/metrics.zig
index e4fd18f1..3585ba6b 100644
--- a/src/signals/datadog/metrics.zig
+++ b/src/signals/datadog/metrics.zig
@@ -201,8 +201,14 @@ fn getDatadogMetricTypeString(series: *MetricSeries) ?[]const u8 {
 /// MetricAccessor template wiring the Datadog metric value primitive.
 /// Metric mutations aren't part of the policy-zig MetricAccessor interface;
 /// the engine only runs filter decisions on metrics.
+/// Datadog metric fields are string-valued; wrap the byte primitive as
+/// `.string` (v0.5.0 removed the `value` accessor field).
+pub fn metricTypedValue(ctx: *const anyopaque, field: policy.MetricFieldRef) ?policy.TypedValue {
+    return .{ .string = metricValue(ctx, field) orelse return null };
+}
+
 pub const metric_accessor: policy.MetricAccessor = .{
-    .value = metricValue,
+    .typed_value = metricTypedValue,
 };
 
 /// Result of evaluating a single metric series
diff --git a/src/signals/otlp/logs.zig b/src/signals/otlp/logs.zig
index 46084d07..2da85576 100644
--- a/src/signals/otlp/logs.zig
+++ b/src/signals/otlp/logs.zig
@@ -371,7 +371,6 @@ fn idTyped(log_ctx: *const OtlpLogContext, id: []const u8) ?policy.TypedValue {
 
 /// LogAccessor template wiring the OTLP log primitives to a registry.
 pub const log_accessor: LogAccessor = .{
-    .value = logValue,
     .exists = logExists,
     .typed_value = logTypedValue,
     .set = logSet,
diff --git a/src/signals/otlp/metrics.zig b/src/signals/otlp/metrics.zig
index 3c4a8ddf..3e988f10 100644
--- a/src/signals/otlp/metrics.zig
+++ b/src/signals/otlp/metrics.zig
@@ -256,10 +256,18 @@ fn getDatapointAttrs(metric: *const Metric) []const KeyValue {
     };
 }
 
+/// Metric fields are all string-valued, so the typed read wraps the byte
+/// primitive as `.string` — byte-identical to what the engine consumed before
+/// v0.5.0 removed the `value` accessor field.
+// ponytail: string-only; add typed numeric reads if metric `equals`/`gt` on numbers is ever needed.
+pub fn metricTypedValue(ctx: *const anyopaque, field: MetricFieldRef) ?policy.TypedValue {
+    return .{ .string = metricValue(ctx, field) orelse return null };
+}
+
 /// MetricAccessor template wiring the OTLP metric value primitive.
 /// Metric mutations aren't part of the policy-zig MetricAccessor interface.
 pub const metric_accessor: policy.MetricAccessor = .{
-    .value = metricValue,
+    .typed_value = metricTypedValue,
 };
 
 /// Result of filtering metrics in-place
diff --git a/src/signals/otlp/traces.zig b/src/signals/otlp/traces.zig
index c8e65246..37408afb 100644
--- a/src/signals/otlp/traces.zig
+++ b/src/signals/otlp/traces.zig
@@ -371,7 +371,6 @@ pub fn traceSet(ctx: *anyopaque, field: TraceFieldRef, value: []const u8) void {
 
 /// TraceAccessor template wiring the OTLP trace primitives.
 pub const trace_accessor: policy.TraceAccessor = .{
-    .value = traceValue,
     .typed_value = traceTypedValue,
     .set = traceSet,
 };
diff --git a/src/signals/prometheus/field_accessor.zig b/src/signals/prometheus/field_accessor.zig
index c62d4ad8..f9fa8911 100644
--- a/src/signals/prometheus/field_accessor.zig
+++ b/src/signals/prometheus/field_accessor.zig
@@ -18,10 +18,16 @@ pub const MetricFieldRef = policy.MetricFieldRef;
 const MetricField = proto.policy.MetricField;
 const AttributePath = proto.policy.AttributePath;
 
+/// Prometheus fields are string-valued; wrap the byte primitive as `.string`
+/// (v0.5.0 removed the `value` accessor field, `typed_value` is now required).
+pub fn metricTypedValue(ctx: *const anyopaque, field: MetricFieldRef) ?policy.TypedValue {
+    return .{ .string = metricValue(ctx, field) orelse return null };
+}
+
 /// MetricAccessor template for Prometheus. Exposition format is immutable
-/// (lines are forwarded verbatim or dropped), so only `value` is wired.
+/// (lines are forwarded verbatim or dropped), so only the read primitive is wired.
 pub const metric_accessor: policy.MetricAccessor = .{
-    .value = metricValue,
+    .typed_value = metricTypedValue,
 };
 
 /// Context for Prometheus field access.
diff --git a/src/signals/prometheus/streaming_filter.zig b/src/signals/prometheus/streaming_filter.zig
index 543012f7..6b5961e8 100644
--- a/src/signals/prometheus/streaming_filter.zig
+++ b/src/signals/prometheus/streaming_filter.zig
@@ -518,7 +518,7 @@ pub const PolicyStreamingFilter = struct {
             &field_accessor.metric_accessor,
             @ptrCast(&ctx),
             &self.policy_id_buf,
-            .{},
+            .{ .io = self.bus.io },
         );
 
         // Continue means keep, drop means filter out
diff --git a/src/tail/eval_context.zig b/src/tail/eval_context.zig
index de946d58..c6265b91 100644
--- a/src/tail/eval_context.zig
+++ b/src/tail/eval_context.zig
@@ -15,6 +15,12 @@ pub const TailLineContext = struct {
     severity: ?[]const u8 = null,
     attrs: std.ArrayListUnmanaged(TailAttr) = .empty,
 
+    /// Typed read primitive (required since v0.5.0). Tail fields are all
+    /// string-valued, so wrap the byte primitive as `.string`.
+    pub fn logTypedValue(ctx_ptr: *const anyopaque, field: FieldRef) ?policy.TypedValue {
+        return .{ .string = logValue(ctx_ptr, field) orelse return null };
+    }
+
     pub fn logValue(ctx_ptr: *const anyopaque, field: FieldRef) ?[]const u8 {
         const self: *const TailLineContext = @ptrCast(@alignCast(ctx_ptr));
         return switch (field) {
@@ -113,7 +119,7 @@ pub const TailLineContext = struct {
 
 /// LogAccessor template wiring the tail-evaluator primitives.
 pub const log_accessor: LogAccessor = .{
-    .value = TailLineContext.logValue,
+    .typed_value = TailLineContext.logTypedValue,
     .set = TailLineContext.logSet,
     .delete = TailLineContext.logDelete,
     .move = TailLineContext.logMove,

From 060685742ac5fb8d99323249cce6e6a3de84d571 Mon Sep 17 00:00:00 2001
From: jaronoff97 <jacobaronoff45@gmail.com>
Date: Fri, 10 Jul 2026 12:18:23 -0400
Subject: [PATCH 2/4] format

---
 src/signals/datadog/logs.zig | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/signals/datadog/logs.zig b/src/signals/datadog/logs.zig
index e5153696..33cdc16c 100644
--- a/src/signals/datadog/logs.zig
+++ b/src/signals/datadog/logs.zig
@@ -817,7 +817,7 @@ test "evalLogRecord - no policies keeps record without output" {
         &registry,
         noop_bus.eventBus(),
         \\{"status": "info", "message": "test1"}
-        ,
+    ,
         null,
     );
     try std.testing.expectEqual(RecordVerdict.keep, verdict);
@@ -856,7 +856,7 @@ test "evalLogRecord - drop policy drops, non-matching keeps" {
         &registry,
         noop_bus.eventBus(),
         \\{"status": "debug", "message": "debug msg"}
-        ,
+    ,
         null,
     );
     try std.testing.expectEqual(RecordVerdict.drop, dropped);
@@ -869,7 +869,7 @@ test "evalLogRecord - drop policy drops, non-matching keeps" {
         &registry,
         noop_bus.eventBus(),
         \\{"status": "error", "message": "error msg"}
-        ,
+    ,
         null,
     );
     try std.testing.expectEqual(RecordVerdict.keep, kept);
@@ -917,7 +917,7 @@ test "evalLogRecord - transform yields replace with serialized record" {
         &registry,
         noop_bus.eventBus(),
         \\{"message": "test log message", "service": "my-service", "status": "info"}
-        ,
+    ,
         null,
     );
     try std.testing.expect(verdict == .replace);

From 0e241d474bfccd594fa4b61f998fa8b390659fff Mon Sep 17 00:00:00 2001
From: jaronoff97 <jacobaronoff45@gmail.com>
Date: Fri, 10 Jul 2026 15:53:05 -0400
Subject: [PATCH 3/4] fix: types for metrics

---
 src/config/types.zig                      | 38 ++++++++++
 src/signals/datadog/log.zig               | 54 ++++++++++++++
 src/signals/datadog/logs.zig              | 85 ++++++++++++++++++++---
 src/signals/otlp/attributes.zig           | 31 +++++++++
 src/signals/otlp/metrics.zig              | 26 +++++--
 src/signals/prometheus/field_accessor.zig | 46 +++++++++++-
 6 files changed, 264 insertions(+), 16 deletions(-)

diff --git a/src/config/types.zig b/src/config/types.zig
index 3815250d..1ebb70ca 100644
--- a/src/config/types.zig
+++ b/src/config/types.zig
@@ -1,5 +1,8 @@
+const std = @import("std");
 const policy = @import("policy_zig");
 
+const log = std.log.scoped(.config);
+
 pub const ProviderConfig = policy.ProviderConfig;
 pub const ServiceMetadata = policy.ServiceMetadata;
 pub const StringPair = policy.StringPair;
@@ -102,4 +105,39 @@ pub const ProxyConfig = struct {
 
     /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default.
     s3_dump: S3DumpConfig = .{},
+
+    /// Post-load validation hook (called by zonfig). Rejects s3-dump knobs that
+    /// would silently break the runtime: a zero flush interval spins the flush
+    /// task in a tight loop, and a zero batch/backlog cap makes the handler drop
+    /// every record. Only checked when the extension is enabled.
+    pub fn validate(self: *ProxyConfig) !void {
+        const s = self.s3_dump;
+        if (!s.enabled) return;
+        if (s.flush_interval_ms == 0) {
+            log.warn("s3_dump.flush_interval_ms must be > 0", .{});
+            return error.InvalidS3DumpConfig;
+        }
+        if (s.max_batch_bytes == 0 or s.max_sealed_bytes == 0 or s.max_batch_records == 0) {
+            log.warn("s3_dump batch caps (max_batch_bytes/max_sealed_bytes/max_batch_records) must be > 0", .{});
+            return error.InvalidS3DumpConfig;
+        }
+    }
 };
+
+test "ProxyConfig.validate rejects zero s3_dump knobs when enabled" {
+    // Disabled: knobs are ignored even if degenerate.
+    var disabled: ProxyConfig = .{ .s3_dump = .{ .enabled = false, .flush_interval_ms = 0 } };
+    try disabled.validate();
+
+    // Enabled with a zero flush interval → tight-spin guard.
+    var zero_interval: ProxyConfig = .{ .s3_dump = .{ .enabled = true, .flush_interval_ms = 0 } };
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_interval.validate());
+
+    // Enabled with a zero batch cap → every record would drop.
+    var zero_bytes: ProxyConfig = .{ .s3_dump = .{ .enabled = true, .max_batch_bytes = 0 } };
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_bytes.validate());
+
+    // Enabled with sane defaults → accepted.
+    var ok: ProxyConfig = .{ .s3_dump = .{ .enabled = true } };
+    try ok.validate();
+}
diff --git a/src/signals/datadog/log.zig b/src/signals/datadog/log.zig
index eedb6455..8fd8c847 100644
--- a/src/signals/datadog/log.zig
+++ b/src/signals/datadog/log.zig
@@ -1,6 +1,7 @@
 const std = @import("std");
 const zimdjson = @import("zimdjson");
 const jscan = @import("../json_scan.zig");
+const policy = @import("policy_zig");
 
 pub const Parser = zimdjson.ondemand.FullParser(.default);
 pub const Value = Parser.Value;
@@ -190,6 +191,59 @@ pub const DatadogLog = struct {
         return jscan.stringSpan(allocator, span) catch null;
     }
 
+    /// Native typed view of a single-segment extra attribute, so the typed
+    /// matchers fire on numeric/boolean values. Handles both storage forms:
+    /// the materializing `parse` path fills `extra` (typed `AnyValue`), while
+    /// the fast `parseRaw` path fills `extra_spans` (raw JSON value text, which
+    /// we classify by first byte). Nested/dotted paths and objects/arrays fall
+    /// back to the string primitive at the call site.
+    pub fn findExtraTyped(self: *const DatadogLog, allocator: std.mem.Allocator, key: []const u8) ?policy.TypedValue {
+        // The two maps are mutually exclusive per parse path (parseRaw fills
+        // `extra_spans`, parse fills `extra`), so probe only the populated one.
+        if (self.extra.count() != 0) {
+            if (self.extra.get(key)) |v| return anyValueTyped(v);
+        } else if (self.extra_spans.get(key)) |span| {
+            return spanTyped(allocator, span);
+        }
+        return null;
+    }
+
+    fn anyValueTyped(v: AnyValue) ?policy.TypedValue {
+        return switch (v) {
+            .bool => |b| .{ .bool = b },
+            .number => |n| switch (n) {
+                .signed => |i| .{ .int = i },
+                // u64 that overflows i64 can't be an .int; widen to double.
+                .unsigned => |u| if (u <= std.math.maxInt(i64))
+                    .{ .int = @intCast(u) }
+                else
+                    .{ .double = @floatFromInt(u) },
+                .double => |d| .{ .double = d },
+            },
+            .string => |s| .{ .string = s.get() catch return null },
+            .null, .object, .array => null,
+        };
+    }
+
+    /// Classify a raw JSON value span (`parseRaw` storage) by its first byte.
+    fn spanTyped(allocator: std.mem.Allocator, span: []const u8) ?policy.TypedValue {
+        if (span.len == 0) return null;
+        switch (span[0]) {
+            '"' => return .{ .string = spanString(allocator, span) orelse return null },
+            't' => return .{ .bool = true },
+            'f' => return .{ .bool = false },
+            'n' => return null, // JSON null: not scalar-matchable
+            else => {
+                // A JSON number: integer if it has no fraction/exponent, else double.
+                if (std.mem.indexOfAny(u8, span, ".eE") == null) {
+                    if (std.fmt.parseInt(i64, span, 10)) |i| return .{ .int = i } else |_| {}
+                }
+                if (std.fmt.parseFloat(f64, span)) |d| return .{ .double = d } else |_| {}
+                return null;
+            },
+        }
+    }
+
     /// Custom JSON serialization for known fields only.
     /// Note: Extra fields are serialized via AnyValue while parser data is alive.
     pub fn jsonStringify(self: *const DatadogLog, jws: *std.json.Stringify) !void {
diff --git a/src/signals/datadog/logs.zig b/src/signals/datadog/logs.zig
index 33cdc16c..0aa4ccb9 100644
--- a/src/signals/datadog/logs.zig
+++ b/src/signals/datadog/logs.zig
@@ -61,6 +61,7 @@ pub fn processLogsStream(
     in_reader: *std.Io.Reader,
     out_writer: *std.Io.Writer,
     content_type: []const u8,
+    sink: ?policy.ExtensionSink,
 ) !StreamProcessResult {
     if (std.mem.indexOf(u8, content_type, "application/json") == null) {
         try streamAll(in_reader, out_writer);
@@ -78,7 +79,7 @@ pub fn processLogsStream(
     // the per-record streaming path reuses one via processLogsSlice instead.
     var parser: Parser = .init;
     defer parser.deinit(allocator);
-    return processLogsSlice(allocator, &parser, allocator, registry, bus, data, out_writer);
+    return processLogsSlice(allocator, &parser, allocator, registry, bus, data, out_writer, sink);
 }
 
 /// Filters a JSON log body already sitting in a slice, reusing a caller-owned
@@ -95,8 +96,9 @@ pub fn processLogsSlice(
     bus: *EventBus,
     data: []const u8,
     out_writer: *std.Io.Writer,
+    sink: ?policy.ExtensionSink,
 ) !StreamProcessResult {
-    const result = try processJsonLogsWithFilter(scratch, parser, parser_gpa, registry, bus, data);
+    const result = try processJsonLogsWithFilter(scratch, parser, parser_gpa, registry, bus, data, sink);
     defer scratch.free(result.data);
     try out_writer.writeAll(result.data);
 
@@ -207,11 +209,23 @@ fn lookupLogAttribute(log: *DatadogLog, allocator: std.mem.Allocator, path: []co
 /// Datadog logs have fields at the root level: message, status/level, ddtags, service, etc.
 /// All attribute types (log, resource, scope) search the same flat namespace since
 /// Datadog has no OTLP-style resource/scope hierarchy.
-/// Typed read primitive (required since v0.5.0). Datadog log fields are all
-/// string-valued in the JSON, so wrap the byte primitive as `.string` —
-/// byte-identical to the pre-v0.5.0 `value` path.
+/// Typed read primitive (required since v0.5.0). Known top-level fields are
+/// strings, but single-segment attributes carry their native JSON type, so the
+/// typed matchers fire on numeric/bool attributes; everything else (nested
+/// attributes, known fields) falls back to the string primitive.
 pub fn logTypedValue(ctx: *const anyopaque, field: FieldRef) ?policy.TypedValue {
-    return .{ .string = logValue(ctx, field) orelse return null };
+    switch (field) {
+        .log_attribute, .resource_attribute, .scope_attribute => |attr_path| {
+            if (attr_path.path.items.len == 1) {
+                const field_ctx: *const FieldAccessorContext = @ptrCast(@alignCast(ctx));
+                if (field_ctx.log.findExtraTyped(field_ctx.allocator, attr_path.path.items[0])) |typed| {
+                    return typed;
+                }
+            }
+            return .{ .string = logValue(ctx, field) orelse return null };
+        },
+        else => return .{ .string = logValue(ctx, field) orelse return null },
+    }
 }
 
 pub fn logValue(ctx: *const anyopaque, field: FieldRef) ?[]const u8 {
@@ -452,6 +466,7 @@ fn processJsonLogsWithFilter(
     registry: *const PolicyRegistry,
     bus: *EventBus,
     data: []const u8,
+    sink: ?policy.ExtensionSink,
 ) !ProcessResult {
     const document = parser.parseFromSlice(parser_gpa, data) catch {
         return returnUnchanged(allocator, data, 0);
@@ -481,7 +496,7 @@ fn processJsonLogsWithFilter(
                 };
 
                 state.original_count += 1;
-                const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, null);
+                const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, sink);
                 if (filter_result.mutated) state.mutated = true;
                 if (filter_result.keep) {
                     try state.kept.append(arena, log_obj);
@@ -498,7 +513,7 @@ fn processJsonLogsWithFilter(
             };
 
             state.original_count = 1;
-            const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, null);
+            const filter_result = filterLog(&engine, &log_obj, arena, &policy_id_buf, sink);
             if (filter_result.mutated) state.mutated = true;
             if (filter_result.keep) {
                 try state.kept.append(arena, log_obj);
@@ -575,6 +590,45 @@ test "datadogFieldAccessor - extra field lookup" {
     try std.testing.expectEqualStrings("abc123-def456", trace_val.?);
 }
 
+test "datadogFieldAccessor - typed attributes are numeric/bool, not string" {
+    const allocator = std.testing.allocator;
+    // No escapes in string values, so the string read borrows (no leak).
+    const json =
+        \\{"message":"x","duration":1234,"ratio":0.5,"ok":true,"name":"svc"}
+    ;
+
+    // Both parse paths must type identically: the materializing `parse` fills
+    // the typed `extra` map; the fast `parseRaw` fills `extra_spans` (raw text).
+    inline for (.{ "parse", "parseRaw" }) |which| {
+        var parser: Parser = .init;
+        defer parser.deinit(allocator);
+
+        var log = if (comptime std.mem.eql(u8, which, "parse")) blk: {
+            const doc = try parser.parseFromSlice(allocator, json);
+            break :blk try DatadogLog.parse(allocator, doc.asValue());
+        } else try DatadogLog.parseRaw(allocator, json);
+        defer log.deinit(allocator);
+
+        var ctx: FieldAccessorContext = .{ .log = &log, .allocator = allocator };
+
+        const duration = logTypedValue(&ctx, .{ .log_attribute = testAttrPath("duration") });
+        try std.testing.expectEqual(@as(i64, 1234), duration.?.int);
+
+        const ratio = logTypedValue(&ctx, .{ .log_attribute = testAttrPath("ratio") });
+        try std.testing.expectEqual(@as(f64, 0.5), ratio.?.double);
+
+        const ok = logTypedValue(&ctx, .{ .log_attribute = testAttrPath("ok") });
+        try std.testing.expectEqual(true, ok.?.bool);
+
+        const name = logTypedValue(&ctx, .{ .log_attribute = testAttrPath("name") });
+        try std.testing.expectEqualStrings("svc", name.?.string);
+
+        // Known top-level fields stay string.
+        const body = logTypedValue(&ctx, .{ .log_field = .LOG_FIELD_BODY });
+        try std.testing.expect(body.? == .string);
+    }
+}
+
 test "datadogFieldAccessor - unwraps JSON-stringified message for body and attributes" {
     const allocator = std.testing.allocator;
 
@@ -969,6 +1023,7 @@ test "processLogs - no policies keeps all logs in array" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1024,6 +1079,7 @@ test "processLogs - DROP policy filters logs from array" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1079,6 +1135,7 @@ test "processLogs - DROP policy drops single object" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1113,6 +1170,7 @@ test "processLogs - malformed JSON returns unchanged (fail-open)" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1146,6 +1204,7 @@ test "processLogs - non-JSON content type returns unchanged" {
         &in_reader,
         &out_writer.writer,
         "text/plain",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1182,6 +1241,7 @@ test "processLogs - Datadog format with ddtags and service" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1239,6 +1299,7 @@ test "processLogs - filter on arbitrary custom field" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1281,6 +1342,7 @@ test "processLogs - extra fields are preserved when no logs dropped" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1342,6 +1404,7 @@ test "processLogs - nested extra fields are preserved when reserializing after d
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1409,6 +1472,7 @@ test "processLogs - nested extra transform is ignored and payload is unchanged"
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1474,6 +1538,7 @@ test "processLogs - mutation triggers reserialization and removes field" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1542,6 +1607,7 @@ test "processLogs - filter on dynamic extra field not in schema" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1614,6 +1680,7 @@ test "processLogs - drops wrapped GCP log via body + nested event_type exact mat
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1687,6 +1754,7 @@ test "processLogs - rewrites fields inside a JSON-wrapped message and re-seriali
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
@@ -1751,6 +1819,7 @@ test "processLogs - filter on nested extra field with exists" {
         &in_reader,
         &out_writer.writer,
         "application/json",
+        null,
     );
     const result: ProcessResult = .{
         .data = try out_writer.toOwnedSlice(),
diff --git a/src/signals/otlp/attributes.zig b/src/signals/otlp/attributes.zig
index 68e9d1a3..d6418b0a 100644
--- a/src/signals/otlp/attributes.zig
+++ b/src/signals/otlp/attributes.zig
@@ -329,6 +329,37 @@ test "getStringValue - empty string" {
     try std.testing.expectEqualStrings("", getStringValue(v).?);
 }
 
+// ── anyValueTyped / findNestedAttributeTyped ──────────
+// These back the typed read primitive for all three OTLP signals (logs,
+// traces, metrics), so the typed matchers (equals/gt/gte/lt/lte) fire on
+// non-string attribute values instead of silently missing them.
+
+test "anyValueTyped - each variant maps to its native type" {
+    try std.testing.expectEqual(@as(i64, 42), anyValueTyped(.{ .value = .{ .int_value = 42 } }).?.int);
+    try std.testing.expectEqual(@as(f64, 0.5), anyValueTyped(.{ .value = .{ .double_value = 0.5 } }).?.double);
+    try std.testing.expectEqual(true, anyValueTyped(.{ .value = .{ .bool_value = true } }).?.bool);
+    try std.testing.expectEqualStrings("hi", anyValueTyped(.{ .value = .{ .string_value = "hi" } }).?.string);
+    try std.testing.expectEqualStrings("\x01\x02", anyValueTyped(.{ .value = .{ .bytes_value = "\x01\x02" } }).?.bytes);
+    try std.testing.expectEqual(@as(?TypedValue, null), anyValueTyped(null));
+    try std.testing.expectEqual(@as(?TypedValue, null), anyValueTyped(.{ .value = null }));
+}
+
+test "findNestedAttributeTyped - typed leaf and nested walk" {
+    const attrs = [_]KeyValue{
+        .{ .key = "code", .value = .{ .value = .{ .int_value = 200 } } },
+        .{ .key = "ok", .value = .{ .value = .{ .bool_value = true } } },
+    };
+    try std.testing.expectEqual(@as(i64, 200), findNestedAttributeTyped(&attrs, &.{"code"}).?.int);
+    try std.testing.expectEqual(true, findNestedAttributeTyped(&attrs, &.{"ok"}).?.bool);
+    try std.testing.expectEqual(@as(?TypedValue, null), findNestedAttributeTyped(&attrs, &.{"missing"}));
+
+    const inner = @constCast(&[_]KeyValue{.{ .key = "status", .value = .{ .value = .{ .int_value = 404 } } }});
+    const nested = [_]KeyValue{.{ .key = "http", .value = .{ .value = .{ .kvlist_value = .{
+        .values = .{ .items = inner, .capacity = inner.len },
+    } } } }};
+    try std.testing.expectEqual(@as(i64, 404), findNestedAttributeTyped(&nested, &.{ "http", "status" }).?.int);
+}
+
 // ── findAttribute ──────────
 
 test "findAttribute - found" {
diff --git a/src/signals/otlp/metrics.zig b/src/signals/otlp/metrics.zig
index 3e988f10..20d260a2 100644
--- a/src/signals/otlp/metrics.zig
+++ b/src/signals/otlp/metrics.zig
@@ -256,12 +256,28 @@ fn getDatapointAttrs(metric: *const Metric) []const KeyValue {
     };
 }
 
-/// Metric fields are all string-valued, so the typed read wraps the byte
-/// primitive as `.string` — byte-identical to what the engine consumed before
-/// v0.5.0 removed the `value` accessor field.
-// ponytail: string-only; add typed numeric reads if metric `equals`/`gt` on numbers is ever needed.
+/// Typed read primitive (required since v0.5.0). Attribute fields carry their
+/// native OTLP type (bool/int/double/bytes/string) so the typed matchers
+/// (`equals`/`gt`/`gte`/`lt`/`lte`) fire on non-string values; the scalar
+/// metric fields (name/unit/schema urls, and the enum tag-name pseudo-fields)
+/// are genuinely strings.
 pub fn metricTypedValue(ctx: *const anyopaque, field: MetricFieldRef) ?policy.TypedValue {
-    return .{ .string = metricValue(ctx, field) orelse return null };
+    const metric_ctx: *const OtlpMetricContext = @ptrCast(@alignCast(ctx));
+    return switch (field) {
+        .datapoint_attribute => |attr_path| otlp_attr.findNestedAttributeTyped(
+            metric_ctx.datapoint_attributes,
+            attr_path.path.items,
+        ),
+        .resource_attribute => |attr_path| if (metric_ctx.resource_metrics.resource) |res|
+            otlp_attr.findNestedAttributeTyped(res.attributes.items, attr_path.path.items)
+        else
+            null,
+        .scope_attribute => |attr_path| if (metric_ctx.scope_metrics.scope) |scope|
+            otlp_attr.findNestedAttributeTyped(scope.attributes.items, attr_path.path.items)
+        else
+            null,
+        else => otlp_attr.typedStr(metricValue(ctx, field)),
+    };
 }
 
 /// MetricAccessor template wiring the OTLP metric value primitive.
diff --git a/src/signals/prometheus/field_accessor.zig b/src/signals/prometheus/field_accessor.zig
index f9fa8911..fbb96e11 100644
--- a/src/signals/prometheus/field_accessor.zig
+++ b/src/signals/prometheus/field_accessor.zig
@@ -18,10 +18,29 @@ pub const MetricFieldRef = policy.MetricFieldRef;
 const MetricField = proto.policy.MetricField;
 const AttributePath = proto.policy.AttributePath;
 
-/// Prometheus fields are string-valued; wrap the byte primitive as `.string`
-/// (v0.5.0 removed the `value` accessor field, `typed_value` is now required).
+/// Typed read primitive (required since v0.5.0). Prometheus labels and metadata
+/// are strings, but the synthetic `value` and `timestamp` datapoint attributes
+/// are numeric — return them as `.double`/`.int` so the typed matchers
+/// (`equals`/`gt`/`gte`/`lt`/`lte`) work (e.g. `datapoint_attribute("value") > 0.5`).
 pub fn metricTypedValue(ctx: *const anyopaque, field: MetricFieldRef) ?policy.TypedValue {
-    return .{ .string = metricValue(ctx, field) orelse return null };
+    switch (field) {
+        .datapoint_attribute => |attr_path| {
+            const key = if (attr_path.path.items.len > 0) attr_path.path.items[0] else return null;
+            const prom_ctx: *const PrometheusFieldContext = @ptrCast(@alignCast(ctx));
+            if (std.mem.eql(u8, key, "value")) {
+                const s = prom_ctx.getValue() orelse return null;
+                // NaN/±Inf parse via parseFloat; anything else falls back to string.
+                return if (std.fmt.parseFloat(f64, s)) |f| .{ .double = f } else |_| .{ .string = s };
+            }
+            if (std.mem.eql(u8, key, "timestamp")) {
+                const s = prom_ctx.getTimestamp() orelse return null;
+                return if (std.fmt.parseInt(i64, s, 10)) |n| .{ .int = n } else |_| .{ .string = s };
+            }
+            // Labels and the "labels" pseudo-field are strings.
+            return .{ .string = metricValue(ctx, field) orelse return null };
+        },
+        else => return .{ .string = metricValue(ctx, field) orelse return null },
+    }
 }
 
 /// MetricAccessor template for Prometheus. Exposition format is immutable
@@ -245,6 +264,27 @@ test "prometheusFieldAccessor - value access" {
     try std.testing.expectEqualStrings("0.75", value.?);
 }
 
+test "prometheusFieldAccessor - typed value/timestamp are numeric, labels are string" {
+    const line = "cpu_usage{host=\"a\"} 0.75 1234567890";
+    const parsed = line_parser.parseLine(line);
+    var ctx = PrometheusFieldContext.fromSample(parsed, line).?;
+
+    // The sample value must be a double so numeric matchers (gt/lt/equals) fire.
+    const value = metricTypedValue(&ctx, .{ .datapoint_attribute = testAttrPath("value") });
+    try std.testing.expect(value.? == .double);
+    try std.testing.expectEqual(@as(f64, 0.75), value.?.double);
+
+    // Timestamp is an integer.
+    const ts = metricTypedValue(&ctx, .{ .datapoint_attribute = testAttrPath("timestamp") });
+    try std.testing.expect(ts.? == .int);
+    try std.testing.expectEqual(@as(i64, 1234567890), ts.?.int);
+
+    // Labels stay string.
+    const host = metricTypedValue(&ctx, .{ .datapoint_attribute = testAttrPath("host") });
+    try std.testing.expect(host.? == .string);
+    try std.testing.expectEqualStrings("a", host.?.string);
+}
+
 test "prometheusFieldAccessor - timestamp access" {
     const line = "cpu_usage 0.75 1234567890";
     const parsed = line_parser.parseLine(line);

From 98d92784e57438535742116ca40c96c9f90dbe64 Mon Sep 17 00:00:00 2001
From: jaronoff97 <jacobaronoff45@gmail.com>
Date: Fri, 10 Jul 2026 16:18:17 -0400
Subject: [PATCH 4/4] comment

---
 src/runtime/extensions.zig   | 7 +++++++
 src/signals/datadog/logs.zig | 8 +++++++-
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/src/runtime/extensions.zig b/src/runtime/extensions.zig
index a044bfb9..569996e4 100644
--- a/src/runtime/extensions.zig
+++ b/src/runtime/extensions.zig
@@ -28,6 +28,13 @@ const log = std.log.scoped(.s3_dump);
 /// accessor context (`*FieldAccessorContext`) borrowed for the duration of
 /// this call; we render its log as one JSON object (the handler appends the
 /// newline that makes the batch ndjson).
+///
+/// The engine dispatches this after keep and before transforms, so the record
+/// is the PRE-transform snapshot by design — for a wrapped-message mutation the
+/// dump carries the original message, not the transformed one the pipeline
+/// forwards downstream. That's the spec's intent (dumps are the record as it
+/// arrived; `mode: dropped` records aren't forwarded at all). See `filterLog`
+/// in signals/datadog/logs.zig.
 pub fn datadogLogEncode(record: *const anyopaque, writer: *std.Io.Writer) anyerror!void {
     const field_ctx: *const dd_logs.FieldAccessorContext = @ptrCast(@alignCast(record));
     try std.json.Stringify.value(field_ctx.log.*, .{}, writer);
diff --git a/src/signals/datadog/logs.zig b/src/signals/datadog/logs.zig
index 0aa4ccb9..a9351142 100644
--- a/src/signals/datadog/logs.zig
+++ b/src/signals/datadog/logs.zig
@@ -370,7 +370,13 @@ fn filterLog(
         policy_id_buf,
         .{ .scratch = allocator, .io = engine.bus.io, .extension_sink = sink },
     );
-    // Re-serialize the wrapped message once if a transform edited inside it.
+    // The extension sink (s3-dump) fires INSIDE evaluate — after keep, before
+    // transforms — so it snapshots the pre-transform record by design (policy
+    // spec v1.6.0; the flagship `mode: dropped` dumps records discarded
+    // downstream anyway). `finalizeWrapped` below rewrites a transform-mutated
+    // wrapped `message` only for the forwarded output; the dump intentionally
+    // keeps the original message. The dispatch point is owned by the engine,
+    // not us, so pre-transform is the only possible (and correct) ordering.
     log.finalizeWrapped(allocator);
     return .{
         .keep = result.decision.shouldContinue(),