feat: end-to-end truncation

AGENTS.md

@@ -30,6 +30,7 @@ go build -tags libsqlite3 ./go/bindings
 # Regenerate checked-in protobuf (required after .proto changes)
 mise run build:go-protobufs
 mise run build:rust-protobufs
+cargo fmt --all  # remove format-only churn from codegen
 
 # Run pgTAP SQL Tests
 mise run ci:sql-tap

crates/dekaf/src/topology.rs

@@ -261,7 +261,7 @@ impl Collection {
 
         let key_schema = avro::key_to_avro(&key_ptr, collection_schema_shape);
 
-        let (not_before, not_after) = (
+        let (mut not_before, not_after) = (
             binding.not_before.map(|b| {
                 uuid::Clock::from_unix(b.seconds.try_into().unwrap(), b.nanos.try_into().unwrap())
             }),
@@ -270,6 +270,23 @@ impl Collection {
             }),
         );
 
+        // Honor truncated-at journal labels: if any partition carries a
+        // truncated-at label, use the max across all partitions and the
+        // binding's not_before as the effective not_before.
+        for partition in &partitions {
+            if let Some(truncated_at_str) = partition
+                .spec
+                .labels
+                .as_ref()
+                .and_then(|ls| ls.labels.iter().find(|l| l.name == ::labels::TRUNCATED_AT))
+                .map(|l| &l.value)
+            {
+                let truncated_clock =
+                    uuid::Clock::from_u64(::labels::parse_truncated_at(truncated_at_str)?);
+                not_before = Some(not_before.map_or(truncated_clock, |nb| nb.max(truncated_clock)));
+            }
+        }
+
         tracing::debug!(
             collection_name,
             partitions = partitions.len(),