diff --git a/examples/tracing/src/test/java/io/github/eschizoid/kpipe/DlqTracePropagationIntegrationTest.java b/examples/tracing/src/test/java/io/github/eschizoid/kpipe/DlqTracePropagationIntegrationTest.java
new file mode 100644
index 00000000..dab7d373
--- /dev/null
+++ b/examples/tracing/src/test/java/io/github/eschizoid/kpipe/DlqTracePropagationIntegrationTest.java
@@ -0,0 +1,193 @@
+package io.github.eschizoid.kpipe;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.tracing.otel.OtelTracer;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.kafka.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/// End-to-end trace propagation across the Kafka boundary onto the dead-letter path.
+///
+/// The existing tracing coverage proves the inbound `traceparent` chains the consumer span and
+/// reaches a normal outbound sink. This fills the remaining gap: when the pipeline FAILS, the
+/// record is routed to the DLQ, and the consumer's active span context must be injected onto
+/// the DLQ record so an operator can follow the failure trace from the original producer into
+/// the dead-letter topic.
+///
+/// The chain under test: inbound record carries a known `traceparent` → consumer starts a span
+/// parented off it (span is current on the worker thread) → an operator throws → the consumer
+/// routes the record to the DLQ via its auto-built producer (wired with the same tracer) →
+/// `injectContextInto` writes the active context onto the DLQ record. The DLQ `traceparent`
+/// must therefore reuse the inbound trace id (same trace) but carry a fresh span id (the
+/// consumer span, not the inbound parent). The exported consumer span is also asserted to be
+/// parented off the inbound span and marked errored.
+///
+/// CI-RUN-REQUIRED: boots a real Kafka broker via Testcontainers — Docker must be available;
+/// this will not run in a Docker-less local sandbox.
+@Testcontainers(disabledWithoutDocker = true)
+class DlqTracePropagationIntegrationTest {
+
+  private static final String KAFKA_VERSION = System.getProperty("kafkaVersion", "4.3.0");
+
+  // Known parent: traceparent format is `version-traceId-parentSpanId-flags`.
+  private static final String INBOUND_TRACE_ID = "11111111111111111111111111111111";
+  private static final String INBOUND_SPAN_ID = "2222222222222222";
+  private static final String INBOUND_TRACEPARENT = "00-" + INBOUND_TRACE_ID + "-" + INBOUND_SPAN_ID + "-01";
+
+  @Container
+  static KafkaContainer kafka = new KafkaContainer(
+    DockerImageName.parse("soldevelo/kafka:%s".formatted(KAFKA_VERSION)).asCompatibleSubstituteFor("apache/kafka")
+  ).withStartupAttempts(3);
+
+  @Test
+  void failingRecordCarriesInboundTraceContextOntoDlqRecord() throws Exception {
+    final var inputTopic = "dlq-trace-in-" + UUID.randomUUID().toString().substring(0, 8);
+    final var dlqTopic = "dlq-trace-dlq-" + UUID.randomUUID().toString().substring(0, 8);
+
+    final var spanExporter = InMemorySpanExporter.create();
+    final var sdk = OpenTelemetrySdk.builder()
+      .setTracerProvider(SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(spanExporter)).build())
+      .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
+      .build();
+
+    final var tracer = new OtelTracer(sdk, "dlq-trace-test");
+
+    // A pipeline that always fails — drives every record onto the DLQ path. The DLQ producer
+    // is auto-built by the consumer and wired with this same tracer, so the active consumer
+    // span is injected onto the DLQ record.
+    final var handle = KPipe.json(inputTopic, consumerProps("dlq-trace-group-" + UUID.randomUUID()))
+      .withTracer(tracer)
+      .withDeadLetterTopic(dlqTopic)
+      .pipe(v -> {
+        throw new IllegalStateException("forced failure to exercise DLQ");
+      })
+      .toConsole()
+      .start();
+
+    try {
+      // Send an inbound record carrying the known traceparent.
+      try (final var producer = new KafkaProducer<byte[], byte[]>(producerProps())) {
+        final var payload = """
+          {"id":1,"event":"boom"}""".getBytes(StandardCharsets.UTF_8);
+        producer
+          .send(
+            new ProducerRecord<byte[], byte[]>(
+              inputTopic,
+              null,
+              null,
+              payload,
+              Collections.singletonList(
+                new RecordHeader("traceparent", INBOUND_TRACEPARENT.getBytes(StandardCharsets.UTF_8))
+              )
+            )
+          )
+          .get();
+        producer.flush();
+      }
+
+      // The DLQ record must carry a traceparent.
+      final var dlqTraceparent = pollFirstTraceparent(dlqTopic, Duration.ofSeconds(25));
+      assertNotNull(dlqTraceparent, "DLQ record must carry a traceparent header");
+
+      // Same trace (proves propagation), different span id (proves the consumer span — not
+      // the inbound parent — is what was injected).
+      final var parts = dlqTraceparent.split("-");
+      assertEquals(4, parts.length, "traceparent should have 4 dash-separated fields");
+      assertEquals(INBOUND_TRACE_ID, parts[1], "DLQ trace id must equal inbound trace id");
+      assertNotEquals(INBOUND_SPAN_ID, parts[2], "DLQ span id must NOT be the inbound parent id");
+
+      // The exported consumer span chains off the inbound parent and is marked errored.
+      final var spans = waitForSpans(spanExporter, Duration.ofSeconds(10));
+      final var span = spans.getFirst();
+      final var parentCtx = span.getParentSpanContext();
+      assertTrue(parentCtx.isValid(), "consumer span must have a valid parent context");
+      assertEquals(INBOUND_TRACE_ID, parentCtx.getTraceId(), "consumer span parent trace id must equal inbound");
+      assertEquals(INBOUND_SPAN_ID, parentCtx.getSpanId(), "consumer span parent span id must equal inbound");
+      assertEquals(inputTopic, span.getAttributes().get(AttributeKey.stringKey("messaging.kafka.topic")));
+      // The injected DLQ span id must be the consumer span that processed the failing record.
+      assertEquals(span.getSpanContext().getSpanId(), parts[2], "DLQ span id must be the consumer span id");
+
+      assertTrue(handle.shutdownGracefully(Duration.ofSeconds(5)));
+    } finally {
+      sdk.close();
+    }
+  }
+
+  private static java.util.List<io.opentelemetry.sdk.trace.data.SpanData> waitForSpans(
+    final InMemorySpanExporter exporter,
+    final Duration timeout
+  ) throws InterruptedException {
+    final var deadline = System.nanoTime() + timeout.toNanos();
+    while (System.nanoTime() < deadline) {
+      final var spans = exporter.getFinishedSpanItems();
+      if (!spans.isEmpty()) return spans;
+      TimeUnit.MILLISECONDS.sleep(100);
+    }
+    throw new AssertionError("Timed out waiting for the consumer span to be exported");
+  }
+
+  /// Polls the DLQ topic until the first record arrives, returns its `traceparent` header value
+  /// (or null if the record had none).
+  private String pollFirstTraceparent(final String topic, final Duration timeout) {
+    try (
+      final var consumer = new KafkaConsumer<byte[], byte[]>(consumerProps("dlq-trace-reader-" + UUID.randomUUID()))
+    ) {
+      consumer.subscribe(Collections.singletonList(topic));
+      final var deadline = System.nanoTime() + timeout.toNanos();
+      while (System.nanoTime() < deadline) {
+        final var records = consumer.poll(Duration.ofMillis(500));
+        for (final var record : records) {
+          final var header = record.headers().lastHeader("traceparent");
+          return header == null ? null : new String(header.value(), StandardCharsets.UTF_8);
+        }
+      }
+    }
+    throw new AssertionError("Timed out waiting for a DLQ record on " + topic);
+  }
+
+  private static Properties consumerProps(final String groupId) {
+    final var props = new Properties();
+    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+    return props;
+  }
+
+  private static Properties producerProps() {
+    final var props = new Properties();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    return props;
+  }
+}
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java
index eb57403d..08eeffca 100644
--- a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java
@@ -153,7 +153,8 @@ void atLeastOnceSurvivesHardCrashRestart() throws Exception {
 
     // CRASH: stop the manager (no further commit; markOffsetProcessed becomes a no-op) and abandon
     // A's consumer thread without a graceful drain — no shutdownGracefully(). On interrupt A's loop
-    // closes its Kafka consumer in its finally, which leaves the group promptly; B then inherits the
+    // closes its Kafka consumer in its finally, which leaves the group promptly; B then inherits
+    // the
     // partition and resumes from the manually-committed prefix.
     offsetManagerA.get().stop();
     threadA.interrupt();
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeRealBrokerBackpressureIntegrationTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeRealBrokerBackpressureIntegrationTest.java
new file mode 100644
index 00000000..621d7631
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeRealBrokerBackpressureIntegrationTest.java
@@ -0,0 +1,227 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.kafka.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/// End-to-end backpressure hysteresis against a real broker, with a no-loss guarantee.
+///
+/// `WatermarkHysteresisTest` pins the pure threshold math, and
+/// `KPipeBackpressureIntegrationTest` drives the full consumer loop against a `MockConsumer`.
+/// Neither exercises a real broker: a MockConsumer's `pause()` is in-memory bookkeeping with
+/// no fetcher behind it, so it cannot show that a paused consumer actually stops pulling bytes
+/// off a live broker and then resumes and drains the backlog without dropping records.
+///
+/// This test fills that gap. A high inbound burst is produced to a real Kafka broker, the sink
+/// is deliberately slow so the in-flight count climbs past the high watermark, and the consumer
+/// is configured with tight custom watermarks in PARALLEL mode. The assertions are:
+///
+///   * **Pause fires.** The consumer pauses at least once — the backpressure pause counter
+///     is positive, proving the in-flight count crossed the high watermark against real
+///     fetch pressure.
+///   * **Resume happens — no permanent stall.** The consumer is not left paused once the sink
+///     drains; the backlog clears and the partition is unpaused.
+///   * **No loss across the pause/resume cycle.** Every produced value is observed exactly
+///     once by the sink (the consumer is the only one in its group, so at-least-once collapses
+///     to exactly the produced set here). A record dropped while paused — or one never
+///     re-fetched after resume — would make the observed set smaller than the produced set.
+///
+/// CI-RUN-REQUIRED: this is a Testcontainers test and needs a Docker daemon to start the Kafka
+/// broker. It compiles locally but cannot run where Docker is unavailable; it runs in CI.
+@Testcontainers(disabledWithoutDocker = true)
+class KPipeRealBrokerBackpressureIntegrationTest {
+
+  private static final System.Logger LOGGER = System.getLogger(
+    KPipeRealBrokerBackpressureIntegrationTest.class.getName()
+  );
+
+  private static final String KAFKA_VERSION = System.getProperty("kafkaVersion", "4.3.0");
+
+  private static final int PARTITIONS = 3;
+  // A burst large enough that the slow sink keeps the in-flight count over the high watermark
+  // long enough for the backpressure loop to observe it and pause, but small enough to drain
+  // inside the test window.
+  private static final int RECORD_COUNT = 600;
+
+  // Tight watermarks so the burst trips the pause early. The hysteresis gap (high > low) is
+  // what stops pause/resume thrashing.
+  private static final long HIGH_WATERMARK = 50L;
+  private static final long LOW_WATERMARK = 10L;
+
+  @Container
+  static KafkaContainer kafka = new KafkaContainer(
+    DockerImageName.parse("soldevelo/kafka:%s".formatted(KAFKA_VERSION)).asCompatibleSubstituteFor("apache/kafka")
+  ).withStartupAttempts(3);
+
+  @Test
+  void parallelBackpressurePausesAtHighWatermarkAndResumesWithoutLoss() throws Exception {
+    final var topic = "bp-real-" + System.nanoTime();
+    final var groupId = "bp-real-group-" + System.nanoTime();
+
+    createTopic(topic);
+    final var producedValues = produceRecords(topic);
+
+    final var observed = ConcurrentHashMap.<String>newKeySet();
+    final var duplicateObservations = ConcurrentHashMap.<String>newKeySet();
+
+    final var consumer = KPipeConsumer.<byte[]>builder()
+      .withProperties(consumerProperties(groupId))
+      .withTopic(topic)
+      .withProcessingMode(ProcessingMode.PARALLEL)
+      // A slow sink so many records are simultaneously in-flight, driving the in-flight count
+      // over the high watermark and forcing a real pause.
+      .withPipeline(
+        TestPipelines.sideEffect(value -> {
+          try {
+            TimeUnit.MILLISECONDS.sleep(15);
+          } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
+          final var asString = new String(value);
+          // Track duplicates so the no-loss check can tell coverage from re-delivery.
+          if (!observed.add(asString)) {
+            duplicateObservations.add(asString);
+          }
+          return value;
+        })
+      )
+      .withBackpressure(HIGH_WATERMARK, LOW_WATERMARK)
+      .build();
+
+    final var thread = Thread.ofVirtual().name("bp-real-consumer").start(consumer::start);
+
+    try {
+      // Pause must fire: the in-flight count crossed the high watermark under real fetch load.
+      awaitCondition(
+        () -> consumer.getMetrics().get(KPipeConsumer.METRIC_BACKPRESSURE_PAUSE_COUNT) > 0,
+        Duration.ofSeconds(30)
+      );
+      final var pauseCount = consumer.getMetrics().get(KPipeConsumer.METRIC_BACKPRESSURE_PAUSE_COUNT);
+      assertTrue(
+        pauseCount > 0,
+        "Consumer must pause at least once under real-broker backpressure; pauses=" + pauseCount
+      );
+
+      // No loss: every produced value is eventually observed. If a record were dropped while
+      // paused or never re-fetched after resume, observed would never reach the produced set.
+      final var allObserved = awaitCondition(() -> observed.containsAll(producedValues), Duration.ofSeconds(90));
+      assertTrue(
+        allObserved,
+        "Every produced record must be observed across the pause/resume cycle; saw %d of %d".formatted(
+          observed.size(),
+          producedValues.size()
+        )
+      );
+      assertEquals(producedValues, observed, "Observed set must exactly cover the produced set (no loss)");
+
+      // Resume must have happened: once the backlog drains the consumer is no longer paused.
+      // The public `inFlight` gauge mirrors the count the backpressure controller reads.
+      final var resumed = awaitCondition(() -> consumer.getMetrics().get("inFlight") == 0L, Duration.ofSeconds(30));
+      assertTrue(resumed, "In-flight count must drain to zero after the burst is processed");
+
+      LOGGER.log(
+        System.Logger.Level.INFO,
+        () ->
+          "real-broker backpressure: pauses=" +
+          pauseCount +
+          " observed=" +
+          observed.size() +
+          " duplicates=" +
+          duplicateObservations.size()
+      );
+    } finally {
+      consumer.close();
+      thread.join(5000);
+    }
+  }
+
+  /// Polls a boolean condition until it holds or the deadline passes. Returns the final state.
+  private boolean awaitCondition(final java.util.function.BooleanSupplier condition, final Duration timeout)
+    throws InterruptedException {
+    final var deadline = System.currentTimeMillis() + timeout.toMillis();
+    while (!condition.getAsBoolean() && System.currentTimeMillis() < deadline) {
+      TimeUnit.MILLISECONDS.sleep(50);
+    }
+    return condition.getAsBoolean();
+  }
+
+  private void createTopic(final String topic) throws Exception {
+    try (final var admin = Admin.create(adminProperties())) {
+      admin.createTopics(List.of(new NewTopic(topic, PARTITIONS, (short) 1))).all().get(30, TimeUnit.SECONDS);
+    }
+  }
+
+  /// Produces a burst of uniquely-valued records spread across all partitions. Returns the
+  /// produced values so the sink can assert exact coverage.
+  private Set<String> produceRecords(final String topic) throws Exception {
+    final var values = ConcurrentHashMap.<String>newKeySet();
+    try (
+      final var producer = new KafkaProducer<>(
+        producerProperties(),
+        new ByteArraySerializer(),
+        new ByteArraySerializer()
+      )
+    ) {
+      final var sends = new ArrayList<Future<?>>();
+      for (int i = 0; i < RECORD_COUNT; i++) {
+        final var value = "val-" + i;
+        values.add(value);
+        final var key = ("key-" + (i % PARTITIONS)).getBytes();
+        sends.add(producer.send(new ProducerRecord<>(topic, key, value.getBytes())));
+      }
+      producer.flush();
+      for (final var send : sends) {
+        send.get(30, TimeUnit.SECONDS);
+      }
+    }
+    return Set.copyOf(values);
+  }
+
+  private Properties adminProperties() {
+    final var props = new Properties();
+    props.put("bootstrap.servers", kafka.getBootstrapServers());
+    return props;
+  }
+
+  private Properties producerProperties() {
+    final var props = new Properties();
+    props.put("bootstrap.servers", kafka.getBootstrapServers());
+    props.put("acks", "all");
+    return props;
+  }
+
+  private Properties consumerProperties(final String groupId) {
+    final var props = new Properties();
+    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+    // A small fetch cap so the burst arrives in many small polls rather than one giant batch.
+    // This gives the in-flight count headroom to climb across watermark crossings instead of
+    // jumping straight over both in a single poll.
+    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");
+    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    return props;
+  }
+}
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KeyOrderedDispatcherCorrectnessTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KeyOrderedDispatcherCorrectnessTest.java
index 0e064332..7222fc14 100644
--- a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KeyOrderedDispatcherCorrectnessTest.java
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KeyOrderedDispatcherCorrectnessTest.java
@@ -280,6 +280,13 @@ void saturationHoldDoesNotLoseOrCorruptRecords() throws InterruptedException {
         () -> "key " + key + " reordered under saturation hold: " + tracker.observed
       );
     }
+    // pendingCount is decremented in the worker's finally, which can run just after the per-record
+    // latch the producers awaited counts down — so poll for the drain rather than reading it the
+    // instant the latch releases (an immediate read flaked on slow CI runners).
+    final var drainDeadline = System.nanoTime() + Duration.ofSeconds(5).toNanos();
+    while (dispatcher.pendingCount() != 0 && System.nanoTime() < drainDeadline) {
+      Thread.onSpinWait();
+    }
     assertEquals(0, dispatcher.pendingCount(), "no records may remain pending after saturation drains");
     dispatcher.close();
   }
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/LifecycleLeakCycleTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/LifecycleLeakCycleTest.java
new file mode 100644
index 00000000..bf011623
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/LifecycleLeakCycleTest.java
@@ -0,0 +1,349 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.sink.BatchPolicy;
+import io.github.eschizoid.kpipe.sink.BatchSink;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+/// Repeated lifecycle stress: build a fresh consumer, start it, then tear it down (clean close or
+/// self-termination crash) many times in a row, asserting nothing accumulates across iterations.
+///
+/// A `KPipeConsumer` is single-use — `start()` does `compareAndSet(CREATED, RUNNING)`, so it cannot
+/// be restarted after close. "Repeated cycles" therefore means a brand-new consumer per iteration;
+/// the leak hazard is that each cycle's resources (the Kafka consumer, the offset manager, the
+/// dispatcher's virtual-thread executor, the scheduler, batch wrappers, metrics-reporter threads)
+/// fail to release, so live threads / open consumers pile up over time.
+///
+/// The probes here are deliberately observable rather than reflective:
+///
+///   * Every iteration uses its own [MockConsumer]; `closed()` proves the Kafka consumer was
+///     released.
+///   * Every iteration uses its own recording [OffsetManager]; its `close()` count proves the
+///     teardown ran exactly once.
+///   * The consumer reaches `CLOSED` every time (`isRunning()` false, `awaitShutdown` true).
+///   * A live-thread census taken before vs. after the whole loop bounds thread growth: the named
+///     kpipe / kafka-consumer threads must not accumulate one-per-iteration. Virtual threads are
+///     unmounted and not individually introspectable, but the scheduler / metrics-reporter are
+///     platform-backed (or named-virtual) and their carrier/daemon threads are what would leak if
+///     `releaseConstructedResources` skipped a shutdown.
+class LifecycleLeakCycleTest {
+
+  private static final String TOPIC = "lifecycle-topic";
+  private static final int CYCLES = 25;
+
+  private Properties props() {
+    final var p = new Properties();
+    p.put("bootstrap.servers", "localhost:9092");
+    p.put("group.id", "lifecycle-group");
+    p.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+    p.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+    p.put("enable.auto.commit", "false");
+    return p;
+  }
+
+  /// Counts its own `close()` invocations and offset marks so a double-release or a skipped release
+  /// is observable. One instance per cycle.
+  private static final class CountingOffsetManager implements OffsetManager<String> {
+
+    final AtomicInteger closeCount = new AtomicInteger(0);
+    final AtomicInteger marks = new AtomicInteger(0);
+
+    @Override
+    public OffsetManager<String> start() {
+      return this;
+    }
+
+    @Override
+    public OffsetManager<String> stop() {
+      return this;
+    }
+
+    @Override
+    public void trackOffset(final ConsumerRecord<String, byte[]> record) {}
+
+    @Override
+    public void markOffsetProcessed(final ConsumerRecord<String, byte[]> record) {
+      marks.incrementAndGet();
+    }
+
+    @Override
+    public void notifyCommitComplete(final String commitId, final boolean success) {}
+
+    @Override
+    public OffsetState getState() {
+      return OffsetState.CREATED;
+    }
+
+    @Override
+    public boolean isRunning() {
+      return true;
+    }
+
+    @Override
+    public Map<String, Object> getStatistics() {
+      return Map.of();
+    }
+
+    @Override
+    public void close() {
+      closeCount.incrementAndGet();
+    }
+  }
+
+  private static MockConsumer<String, byte[]> nonSubscribingMock() {
+    return new MockConsumer<>("earliest") {
+      @Override
+      public synchronized void subscribe(final Collection<String> topics) {}
+
+      @Override
+      public synchronized void subscribe(final Collection<String> topics, final ConsumerRebalanceListener cb) {}
+    };
+  }
+
+  private static MockConsumer<String, byte[]> seeded(final int recordCount) {
+    final var mock = nonSubscribingMock();
+    final var tp = new TopicPartition(TOPIC, 0);
+    mock.assign(List.of(tp));
+    mock.updateBeginningOffsets(Map.of(tp, 0L));
+    for (int i = 0; i < recordCount; i++) {
+      mock.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k" + i, ("v" + i).getBytes()));
+    }
+    mock.updateEndOffsets(Map.of(tp, (long) recordCount));
+    return mock;
+  }
+
+  /// Snapshot of live thread names that the consumer might spawn. Virtual worker threads mount on
+  /// shared carriers and aren't reliably countable; the names below are the ones a leaked teardown
+  /// would strand: the scheduler carrier, the metrics reporter, the named consumer thread.
+  private static Set<String> kpipeThreadNames() {
+    return Thread.getAllStackTraces()
+      .keySet()
+      .stream()
+      .map(Thread::getName)
+      .filter(n -> n.startsWith("kpipe-") || n.startsWith("kafka-consumer-"))
+      .collect(Collectors.toSet());
+  }
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void repeatedBuildStartCloseReleasesEverythingEachCycle(final ProcessingMode mode) throws Exception {
+    // Clean lifecycle: build → start → process a few records → close, repeated. Each cycle must
+    // close its own Kafka consumer and offset manager exactly once and reach CLOSED. A leak would
+    // show up as a mock left open, a manager closed != once, or a consumer stuck non-CLOSED.
+    final var threadsBefore = kpipeThreadNames();
+
+    for (int cycle = 0; cycle < CYCLES; cycle++) {
+      final var mock = seeded(5);
+      final var manager = new CountingOffsetManager();
+      final var processed = new AtomicInteger(0);
+
+      final var consumer = KPipeConsumer.<String>builder()
+        .withProperties(props())
+        .withTopic(TOPIC)
+        .withProcessingMode(mode)
+        .withPipeline(
+          TestPipelines.sideEffect(v -> {
+            processed.incrementAndGet();
+            return v;
+          })
+        )
+        .withConsumer(() -> mock)
+        .withOffsetManager(manager)
+        .withPollTimeout(Duration.ofMillis(5))
+        .withThreadTerminationTimeout(Duration.ofSeconds(2))
+        .build();
+
+      consumer.start();
+      // Don't require all 5 — the point is the consumer is genuinely running before we close.
+      final var deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos();
+      while (processed.get() == 0 && System.nanoTime() < deadline) Thread.sleep(2);
+      consumer.close();
+
+      assertTrue(consumer.awaitShutdown(Duration.ofSeconds(5)), () -> "cycle must reach CLOSED for mode " + mode);
+      assertFalse(consumer.isRunning(), () -> "consumer must not be running after close for mode " + mode);
+      assertTrue(mock.closed(), () -> "cycle must close its Kafka consumer for mode " + mode);
+      assertEquals(
+        1,
+        manager.closeCount.get(),
+        () -> "offset manager must be closed exactly once per clean cycle for mode " + mode
+      );
+    }
+
+    assertThreadsDidNotAccumulate(threadsBefore, mode);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void repeatedBuildStartCrashReleasesEverythingEachCycle(final ProcessingMode mode) throws Exception {
+    // Crash lifecycle: the consumer thread self-terminates because poll() throws an Error that
+    // escapes the loop's catch(Exception). The thread's own finally must still run
+    // releaseConstructedResources — closing the Kafka consumer + offset manager — without any
+    // external close(). Repeated to prove the crash path doesn't leak per iteration.
+    final var threadsBefore = kpipeThreadNames();
+
+    for (int cycle = 0; cycle < CYCLES; cycle++) {
+      final var manager = new CountingOffsetManager();
+      final var poisoned = new AtomicBoolean(false);
+      final var pollEntered = new CountDownLatch(1);
+      final var tp = new TopicPartition(TOPIC, 0);
+
+      final var mock = new MockConsumer<String, byte[]>("earliest") {
+        @Override
+        public synchronized void subscribe(final Collection<String> topics) {}
+
+        @Override
+        public synchronized void subscribe(final Collection<String> topics, final ConsumerRebalanceListener cb) {}
+
+        @Override
+        public synchronized ConsumerRecords<String, byte[]> poll(final Duration timeout) {
+          pollEntered.countDown();
+          if (poisoned.get()) throw new Error("simulated consumer-thread crash on cycle");
+          return ConsumerRecords.empty();
+        }
+      };
+      mock.assign(List.of(tp));
+      mock.updateBeginningOffsets(Map.of(tp, 0L));
+
+      final var consumer = KPipeConsumer.<String>builder()
+        .withProperties(props())
+        .withTopic(TOPIC)
+        .withProcessingMode(mode)
+        .withPipeline(TestPipelines.identity())
+        .withConsumer(() -> mock)
+        .withOffsetManager(manager)
+        .withPollTimeout(Duration.ofMillis(5))
+        .withThreadTerminationTimeout(Duration.ofSeconds(2))
+        .build();
+
+      consumer.start();
+      assertTrue(pollEntered.await(2, TimeUnit.SECONDS), "consumer thread must reach poll() before we poison it");
+      poisoned.set(true);
+
+      // The crash drives teardown from the thread's own finally. Wait for the manager to be
+      // closed — that's the last-but-one step of releaseConstructedResources.
+      final var deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos();
+      while (manager.closeCount.get() == 0 && System.nanoTime() < deadline) Thread.sleep(5);
+
+      assertEquals(
+        1,
+        manager.closeCount.get(),
+        () -> "self-terminating crash must release the offset manager exactly once for mode " + mode
+      );
+      assertTrue(
+        consumer.awaitShutdown(Duration.ofSeconds(5)),
+        () -> "crashed consumer must reach CLOSED for mode " + mode
+      );
+      assertFalse(consumer.isRunning(), () -> "crashed consumer must not be running for mode " + mode);
+
+      // A late external close() after a self-terminating crash must be a clean no-op — not a
+      // second release (which would double-close the manager).
+      consumer.close();
+      assertEquals(
+        1,
+        manager.closeCount.get(),
+        () -> "external close() after a crash must not re-release resources for mode " + mode
+      );
+    }
+
+    assertThreadsDidNotAccumulate(threadsBefore, mode);
+  }
+
+  @Test
+  void repeatedCycleWithFullFeatureStackDoesNotLeak() throws Exception {
+    // Heaviest per-cycle resource footprint: PARALLEL dispatcher executor + scheduler (needed by
+    // both batch wrappers and the circuit breaker) + a batch wrapper + a DLQ producer-less retry
+    // path. If any of these aren't torn down per cycle, the scheduler carrier or batch wrapper
+    // age-tick task would strand threads that the census catches.
+    final var threadsBefore = kpipeThreadNames();
+
+    for (int cycle = 0; cycle < CYCLES; cycle++) {
+      final var mock = seeded(4);
+      final var manager = new CountingOffsetManager();
+      final var flushed = new AtomicInteger(0);
+
+      final var consumer = KPipeConsumer.<String>builder()
+        .withProperties(props())
+        .withBatchPipeline(
+          TOPIC,
+          TestPipelines.sideEffect(v -> v),
+          BatchSink.ofVoid(batch -> flushed.addAndGet(batch.size())),
+          BatchPolicy.ofSize(2)
+        )
+        .withProcessingMode(ProcessingMode.PARALLEL)
+        .withCircuitBreaker(new CircuitBreakerController(0.5, 10, Duration.ofSeconds(30)))
+        .withRetry(1, Duration.ofMillis(1))
+        .withConsumer(() -> mock)
+        .withOffsetManager(manager)
+        .withPollTimeout(Duration.ofMillis(5))
+        .withThreadTerminationTimeout(Duration.ofSeconds(2))
+        .build();
+
+      consumer.start();
+      final var deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos();
+      while (flushed.get() == 0 && System.nanoTime() < deadline) Thread.sleep(2);
+      consumer.close();
+
+      assertTrue(consumer.awaitShutdown(Duration.ofSeconds(5)), "full-stack cycle must reach CLOSED");
+      assertTrue(mock.closed(), "full-stack cycle must close its Kafka consumer");
+      assertEquals(1, manager.closeCount.get(), "full-stack cycle must close the offset manager exactly once");
+    }
+
+    assertThreadsDidNotAccumulate(threadsBefore, ProcessingMode.PARALLEL);
+  }
+
+  /// Bounds thread growth across the whole loop. Newly-stranded kpipe / kafka-consumer threads must
+  /// not number on the order of `CYCLES`. A small slack absorbs late-joining carrier threads and
+  /// JIT/GC daemons that share the prefix-free namespace; a genuine per-cycle leak would blow far
+  /// past it.
+  private static void assertThreadsDidNotAccumulate(final Set<String> before, final ProcessingMode mode)
+    throws InterruptedException {
+    // Give teardown a moment to let any just-interrupted threads actually exit before the census.
+    final var deadline = System.nanoTime() + Duration.ofSeconds(3).toNanos();
+    while (System.nanoTime() < deadline) {
+      final var leaked = leakedSince(before);
+      if (leaked.size() <= 3) break;
+      Thread.sleep(50);
+    }
+    final var leaked = leakedSince(before);
+    assertTrue(
+      leaked.size() <= 3,
+      () ->
+        "mode " +
+        mode +
+        " leaked kpipe/kafka-consumer threads across " +
+        CYCLES +
+        " cycles: " +
+        leaked +
+        " (a per-cycle leak would show ~" +
+        CYCLES +
+        " stranded threads)"
+    );
+  }
+
+  private static Set<String> leakedSince(final Set<String> before) {
+    final var now = kpipeThreadNames();
+    now.removeAll(before);
+    return now;
+  }
+}
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/ProcessingModeSinkDlqMatrixTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/ProcessingModeSinkDlqMatrixTest.java
new file mode 100644
index 00000000..59d184a7
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/ProcessingModeSinkDlqMatrixTest.java
@@ -0,0 +1,494 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.sink.BatchPolicy;
+import io.github.eschizoid.kpipe.sink.BatchResult;
+import io.github.eschizoid.kpipe.sink.BatchSink;
+import io.github.eschizoid.kpipe.sink.CompositeMessageSink;
+import io.github.eschizoid.kpipe.sink.MessageSink;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.Mockito;
+
+/// The combinatorial cell of the lifecycle matrix that isn't already pinned elsewhere:
+/// [ProcessingMode] × sink shape (single / multi / batch) × DLQ on/off × retry × circuit breaker,
+/// each driven end-to-end through a started [KPipeConsumer] against a [MockConsumer] (no Docker).
+/// Every cell asserts the at-least-once invariants:
+///
+///   * **No loss, no silent drop** — every seeded record lands in exactly one terminal bucket
+///     (success sink OR dead-letter), and the two buckets together cover the full seeded
+///     offset set.
+///   * **Offsets advance** — `markOffsetProcessed` is called for every record so the commit offset
+///     advances past the batch (no infinite re-fetch).
+///
+/// Cells deliberately NOT duplicated here (already covered, with the owning test named so a future
+/// reader doesn't re-add them):
+///
+///   * ProcessingMode × ordering / per-key serialization / offset-drain correctness —
+///     `DispatcherIntegrationTest` (all three modes, plus KEY_ORDERED × retry and KEY_ORDERED ×
+///     error-handler failure).
+///   * Circuit-breaker trip / half-open probe / recovery, including the batch-failure trip path —
+///     `KPipeCircuitBreakerIntegrationTest` (SEQUENTIAL + batch). Not re-run per mode here: the
+///     breaker observes pipeline outcomes through the same `recordSuccess` / `recordFailure` hooks
+///     regardless of dispatcher, so a third copy per mode would add cost without a new property.
+///   * DLQ send mechanics via `processRecord` directly (synchronous send, throwing send, null
+///     pipeline, bundled `withDeadLetterQueue` setter) — `KPipeDlqTest`.
+///
+/// Dropped cells (logged here rather than silently truncated): the full 3×3×2×2×2 = 72-cell product
+/// is not enumerated. Retry+CB are toggled together with the "all features" cell rather than as
+/// independent axes because their interaction with the dispatcher is identical to their interaction
+/// already proven in the single-axis suites; the high-value gap is the sink-shape × DLQ × mode
+/// face, which is what this class fills.
+class ProcessingModeSinkDlqMatrixTest {
+
+  private static final String TOPIC = "matrix-topic";
+  private static final String DLQ_TOPIC = "matrix-dlq";
+  private static final int RECORDS = 12;
+
+  private static Properties props() {
+    final var p = new Properties();
+    p.put("bootstrap.servers", "localhost:9092");
+    p.put("group.id", "matrix-group");
+    p.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+    p.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+    p.put("enable.auto.commit", "false");
+    return p;
+  }
+
+  /// Records every `markOffsetProcessed` offset so we can assert the commit offset would advance
+  /// past every seeded record (no re-fetch loop).
+  private static final class MarkRecordingOffsetManager implements OffsetManager<String> {
+
+    final Set<Long> markedOffsets = ConcurrentHashMap.newKeySet();
+
+    @Override
+    public OffsetManager<String> start() {
+      return this;
+    }
+
+    @Override
+    public OffsetManager<String> stop() {
+      return this;
+    }
+
+    @Override
+    public void trackOffset(final ConsumerRecord<String, byte[]> record) {}
+
+    @Override
+    public void markOffsetProcessed(final ConsumerRecord<String, byte[]> record) {
+      markedOffsets.add(record.offset());
+    }
+
+    @Override
+    public void notifyCommitComplete(final String commitId, final boolean success) {}
+
+    @Override
+    public OffsetState getState() {
+      return OffsetState.CREATED;
+    }
+
+    @Override
+    public boolean isRunning() {
+      return true;
+    }
+
+    @Override
+    public Map<String, Object> getStatistics() {
+      return Map.of();
+    }
+
+    @Override
+    public void close() {}
+  }
+
+  private static MockConsumer<String, byte[]> seeded(final int recordCount) {
+    final var mock = new MockConsumer<String, byte[]>("earliest") {
+      @Override
+      public synchronized void subscribe(final Collection<String> topics) {}
+
+      @Override
+      public synchronized void subscribe(final Collection<String> topics, final ConsumerRebalanceListener cb) {}
+    };
+    final var tp = new TopicPartition(TOPIC, 0);
+    mock.assign(List.of(tp));
+    mock.updateBeginningOffsets(Map.of(tp, 0L));
+    for (int i = 0; i < recordCount; i++) {
+      // Value encodes the offset so the pipeline can decide pass/fail per record.
+      mock.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k" + i, ("v" + i).getBytes(StandardCharsets.UTF_8)));
+    }
+    mock.updateEndOffsets(Map.of(tp, (long) recordCount));
+    return mock;
+  }
+
+  private static long offsetOf(final byte[] value) {
+    // Strip the leading "v" and parse the offset the seeder encoded.
+    return Long.parseLong(new String(value, StandardCharsets.UTF_8).substring(1));
+  }
+
+  /// A DLQ producer that records every send into a concurrent list and completes the future, so a
+  /// started consumer's synchronous DLQ send succeeds and we can read back which offsets parked.
+  @SuppressWarnings("unchecked")
+  private static Producer<String, byte[]> recordingDlqProducer(final List<Long> dlqOffsets) {
+    final Producer<String, byte[]> producer = Mockito.mock(Producer.class);
+    Mockito.lenient()
+      .when(producer.send(Mockito.any(ProducerRecord.class)))
+      .thenAnswer(inv -> {
+        final ProducerRecord<String, byte[]> rec = inv.getArgument(0);
+        dlqOffsets.add(offsetOf(rec.value()));
+        return CompletableFuture.completedFuture(Mockito.mock(RecordMetadata.class));
+      });
+    return producer;
+  }
+
+  private static void awaitCondition(final BooleanSupplier cond, final long timeoutMs) throws InterruptedException {
+    final long deadline = System.currentTimeMillis() + timeoutMs;
+    while (!cond.getAsBoolean()) {
+      if (System.currentTimeMillis() >= deadline) throw new AssertionError(
+        "awaitCondition timed out after " + timeoutMs + "ms — invariant assertions would have run on a partial dataset"
+      );
+      Thread.sleep(10);
+    }
+  }
+
+  /// Every offset in `[0, RECORDS)` must appear in exactly one of the two terminal buckets, the two
+  /// buckets must be disjoint, and every offset must have been marked processed. This is the
+  /// at-least-once-without-loss-and-without-duplicate-terminal-bucket invariant for the cell.
+  private static void assertExactlyOnceTerminal(
+    final Collection<Long> succeeded,
+    final Collection<Long> deadLettered,
+    final Set<Long> marked,
+    final String cellLabel
+  ) {
+    final var union = new HashSet<Long>();
+    union.addAll(succeeded);
+    union.addAll(deadLettered);
+
+    final var overlap = new HashSet<>(succeeded);
+    overlap.retainAll(deadLettered);
+    assertTrue(
+      overlap.isEmpty(),
+      () -> cellLabel + ": offsets reached BOTH success and DLQ (double terminal): " + overlap
+    );
+
+    for (long off = 0; off < RECORDS; off++) {
+      final long o = off;
+      assertTrue(
+        union.contains(o),
+        () ->
+          cellLabel +
+          ": offset " +
+          o +
+          " reached NO terminal bucket (silent loss). success=" +
+          succeeded +
+          " dlq=" +
+          deadLettered
+      );
+      assertTrue(
+        marked.contains(o),
+        () -> cellLabel + ": offset " + o + " was never marked processed (would re-fetch forever). marked=" + marked
+      );
+    }
+  }
+
+  // ────────────────────────────────────────────────────────────────────────────────
+  // Cell 1: single sink × all-success × no DLQ — baseline at-least-once per mode
+  // ────────────────────────────────────────────────────────────────────────────────
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void singleSinkAllSuccessMarksEveryOffset(final ProcessingMode mode) throws Exception {
+    final var sinkSaw = new CopyOnWriteArrayList<Long>();
+    final var manager = new MarkRecordingOffsetManager();
+    final var mock = seeded(RECORDS);
+    final MessageSink<byte[]> sink = v -> sinkSaw.add(offsetOf(v));
+
+    final var consumer = KPipeConsumer.<String>builder()
+      .withProperties(props())
+      .withTopic(TOPIC)
+      .withProcessingMode(mode)
+      .withPipeline(
+        TestPipelines.sideEffect(v -> {
+          sink.accept(v);
+          return v;
+        })
+      )
+      .withConsumer(() -> mock)
+      .withOffsetManager(manager)
+      .withPollTimeout(Duration.ofMillis(5))
+      .build();
+
+    try {
+      consumer.start();
+      awaitCondition(() -> sinkSaw.size() >= RECORDS && manager.markedOffsets.size() >= RECORDS, 10_000);
+
+      assertExactlyOnceTerminal(sinkSaw, List.of(), manager.markedOffsets, "single-sink/all-success/" + mode);
+    } finally {
+      consumer.close();
+    }
+  }
+
+  // ────────────────────────────────────────────────────────────────────────────────
+  // Cell 2: multi-sink fanout × all-success — every sink sees every record
+  // ────────────────────────────────────────────────────────────────────────────────
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void multiSinkFanoutEverySinkSeesEveryRecord(final ProcessingMode mode) throws Exception {
+    // Drive both terminal sinks through the real CompositeMessageSink broadcast (the explicit-API
+    // multi-sink shape), not two inline adds — so the assertion reflects the fanout machinery's
+    // delivery, not the test harness. Both sinks must see all records regardless of dispatcher,
+    // and offsets still mark exactly once.
+    final var sinkA = new CopyOnWriteArrayList<Long>();
+    final var sinkB = new CopyOnWriteArrayList<Long>();
+    final MessageSink<byte[]> recordA = v -> sinkA.add(offsetOf(v));
+    final MessageSink<byte[]> recordB = v -> sinkB.add(offsetOf(v));
+    final var fanout = new CompositeMessageSink<>(List.of(recordA, recordB));
+    final var manager = new MarkRecordingOffsetManager();
+    final var mock = seeded(RECORDS);
+
+    final var consumer = KPipeConsumer.<String>builder()
+      .withProperties(props())
+      .withTopic(TOPIC)
+      .withProcessingMode(mode)
+      .withPipeline(
+        TestPipelines.sideEffect(v -> {
+          fanout.accept(v);
+          return v;
+        })
+      )
+      .withConsumer(() -> mock)
+      .withOffsetManager(manager)
+      .withPollTimeout(Duration.ofMillis(5))
+      .build();
+
+    try {
+      consumer.start();
+      awaitCondition(
+        () -> sinkA.size() >= RECORDS && sinkB.size() >= RECORDS && manager.markedOffsets.size() >= RECORDS,
+        10_000
+      );
+
+      assertExactlyOnceTerminal(sinkA, List.of(), manager.markedOffsets, "multi-sink-A/" + mode);
+      assertEquals(
+        new HashSet<>(sinkA),
+        new HashSet<>(sinkB),
+        () -> "both sinks in the fanout must see the same offset set for mode " + mode
+      );
+    } finally {
+      consumer.close();
+    }
+  }
+
+  // ────────────────────────────────────────────────────────────────────────────────
+  // Cell 3: single sink × retry-then-DLQ — failing records park, successes sink
+  // ────────────────────────────────────────────────────────────────────────────────
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void retryThenDlqRoutesFailuresAndKeepsAtLeastOnce(final ProcessingMode mode) throws Exception {
+    // Records whose offset is even fail every attempt → after retries exhaust they must reach the
+    // DLQ. Odd offsets succeed → reach the success sink. The two buckets must exactly partition the
+    // seeded set, every offset must be marked. This is the per-mode DLQ-through-a-started-consumer
+    // gap (KPipeDlqTest only drives processRecord directly on the default mode).
+    final var sinkSaw = new CopyOnWriteArrayList<Long>();
+    final var dlqOffsets = new CopyOnWriteArrayList<Long>();
+    final var manager = new MarkRecordingOffsetManager();
+    final var mock = seeded(RECORDS);
+    final var dlqProducer = recordingDlqProducer(dlqOffsets);
+
+    final var consumer = KPipeConsumer.<String>builder()
+      .withProperties(props())
+      .withTopic(TOPIC)
+      .withProcessingMode(mode)
+      .withPipeline(
+        TestPipelines.sideEffect(v -> {
+          final var off = offsetOf(v);
+          if (off % 2 == 0) throw new RuntimeException("deliberate failure at offset " + off);
+          sinkSaw.add(off);
+          return v;
+        })
+      )
+      .withRetry(1, Duration.ofMillis(1))
+      .withDeadLetterTopic(DLQ_TOPIC)
+      .withKafkaProducer(dlqProducer)
+      .withConsumer(() -> mock)
+      .withOffsetManager(manager)
+      .withPollTimeout(Duration.ofMillis(5))
+      .build();
+
+    try {
+      consumer.start();
+      final var expectedSuccess = RECORDS / 2;
+      final var expectedDlq = RECORDS - expectedSuccess;
+      awaitCondition(
+        () ->
+          sinkSaw.size() >= expectedSuccess &&
+          dlqOffsets.size() >= expectedDlq &&
+          manager.markedOffsets.size() >= RECORDS,
+        10_000
+      );
+
+      assertExactlyOnceTerminal(sinkSaw, dlqOffsets, manager.markedOffsets, "retry-then-dlq/" + mode);
+      // Sanity: the partition matches the even/odd rule we encoded.
+      assertTrue(sinkSaw.stream().allMatch(o -> o % 2 == 1), () -> "only odd offsets should succeed: " + sinkSaw);
+      assertTrue(dlqOffsets.stream().allMatch(o -> o % 2 == 0), () -> "only even offsets should DLQ: " + dlqOffsets);
+    } finally {
+      consumer.close();
+    }
+  }
+
+  // ────────────────────────────────────────────────────────────────────────────────
+  // Cell 4: batch sink × per-record DLQ — partial-batch failure routing per mode
+  // ────────────────────────────────────────────────────────────────────────────────
+
+  @ParameterizedTest
+  @EnumSource(ProcessingMode.class)
+  void batchSinkPartialFailureRoutesPerRecordDlq(final ProcessingMode mode) throws Exception {
+    // A batch sink that fails the even-offset records within each batch via a per-index
+    // BatchResult. The wrapper must route exactly those to the DLQ and mark the succeeded ones,
+    // covering the full seeded set. Exercises the batch coverage-contract path under each
+    // dispatcher. RECORDS=12 with size 5 gives batches [5, 5, 2] — the trailing 2 are a genuine
+    // partial batch that only flushes at teardown, so this also exercises the close()-drain path.
+    final var sinkSaw = new CopyOnWriteArrayList<Long>();
+    final var dlqOffsets = new CopyOnWriteArrayList<Long>();
+    final var manager = new MarkRecordingOffsetManager();
+    final var mock = seeded(RECORDS);
+    final var dlqProducer = recordingDlqProducer(dlqOffsets);
+
+    final BatchSink<byte[]> batchSink = batch -> {
+      final var succeeded = new ArrayList<Integer>();
+      final var failed = new HashMap<Integer, Exception>();
+      for (int i = 0; i < batch.size(); i++) {
+        final var off = offsetOf(batch.get(i));
+        if (off % 2 == 0) {
+          failed.put(i, new RuntimeException("batch failure at offset " + off));
+        } else {
+          succeeded.add(i);
+          sinkSaw.add(off);
+        }
+      }
+      return new BatchResult(succeeded, failed);
+    };
+
+    final var consumer = KPipeConsumer.<String>builder()
+      .withProperties(props())
+      .withBatchPipeline(TOPIC, TestPipelines.sideEffect(v -> v), batchSink, BatchPolicy.ofSize(5))
+      .withProcessingMode(mode)
+      .withDeadLetterTopic(DLQ_TOPIC)
+      .withKafkaProducer(dlqProducer)
+      .withConsumer(() -> mock)
+      .withOffsetManager(manager)
+      .withPollTimeout(Duration.ofMillis(5))
+      .build();
+
+    try {
+      consumer.start();
+      final var expectedSuccess = RECORDS / 2;
+      final var expectedDlq = RECORDS - expectedSuccess;
+      // Two batches flush on size(5) (10 records); the trailing partial batch of 2 flushes at
+      // teardown — so we close() to force the final flush, then assert. Wait for the size-triggered
+      // flushes first to avoid racing the close.
+      awaitCondition(() -> sinkSaw.size() + dlqOffsets.size() >= 10, 10_000);
+    } finally {
+      consumer.close();
+    }
+
+    // After close() the trailing buffered batch has flushed (teardown drain). Now the full set is
+    // terminal.
+    assertExactlyOnceTerminal(sinkSaw, dlqOffsets, manager.markedOffsets, "batch-partial-dlq/" + mode);
+    assertTrue(
+      sinkSaw.stream().allMatch(o -> o % 2 == 1),
+      () -> "only odd offsets should succeed in batch: " + sinkSaw
+    );
+    assertTrue(
+      dlqOffsets.stream().allMatch(o -> o % 2 == 0),
+      () -> "only even offsets should DLQ in batch: " + dlqOffsets
+    );
+  }
+
+  // ────────────────────────────────────────────────────────────────────────────────
+  // Cell 5: the "everything on" cell — PARALLEL + retry + CB + DLQ + single sink
+  // ────────────────────────────────────────────────────────────────────────────────
+
+  @Test
+  void allFeaturesEnabledStillSatisfiesAtLeastOnce() throws Exception {
+    // The densest realistic config: PARALLEL dispatch, retry, a circuit breaker (configured but
+    // never tripping because the failure rate stays below threshold), DLQ for the few failures, a
+    // single success sink. Every record must still reach exactly one terminal bucket and be marked.
+    final var sinkSaw = new CopyOnWriteArrayList<Long>();
+    final var dlqOffsets = new CopyOnWriteArrayList<Long>();
+    final var manager = new MarkRecordingOffsetManager();
+    final var mock = seeded(RECORDS);
+    final var dlqProducer = recordingDlqProducer(dlqOffsets);
+    final var attempts = new ConcurrentHashMap<Long, AtomicInteger>();
+
+    final var consumer = KPipeConsumer.<String>builder()
+      .withProperties(props())
+      .withTopic(TOPIC)
+      .withProcessingMode(ProcessingMode.PARALLEL)
+      // High windowSize/threshold so sparse failures never trip the breaker here.
+      .withCircuitBreaker(new CircuitBreakerController(0.9, 50, Duration.ofSeconds(30)))
+      .withRetry(2, Duration.ofMillis(1))
+      .withPipeline(
+        TestPipelines.sideEffect(v -> {
+          final var off = offsetOf(v);
+          // Offset 3 recovers on retry; offset 7 fails permanently to the DLQ.
+          final var n = attempts.computeIfAbsent(off, _ -> new AtomicInteger(0)).incrementAndGet();
+          if (off == 3 && n == 1) throw new RuntimeException("transient failure, recovers on retry");
+          if (off == 7) throw new RuntimeException("permanent failure → DLQ");
+          sinkSaw.add(off);
+          return v;
+        })
+      )
+      .withDeadLetterTopic(DLQ_TOPIC)
+      .withKafkaProducer(dlqProducer)
+      .withConsumer(() -> mock)
+      .withOffsetManager(manager)
+      .withPollTimeout(Duration.ofMillis(5))
+      .build();
+
+    try {
+      consumer.start();
+      awaitCondition(
+        () -> sinkSaw.size() >= RECORDS - 1 && dlqOffsets.contains(7L) && manager.markedOffsets.size() >= RECORDS,
+        10_000
+      );
+
+      assertExactlyOnceTerminal(sinkSaw, dlqOffsets, manager.markedOffsets, "all-features");
+      assertEquals(List.of(7L), dlqOffsets, "only the permanently-failing offset 7 should reach the DLQ");
+      assertTrue(sinkSaw.contains(3L), "offset 3 must succeed on retry, not DLQ");
+      assertEquals(
+        0L,
+        consumer.getMetrics().get(KPipeConsumer.METRIC_CIRCUIT_BREAKER_TRIPS),
+        "the breaker must not trip at this sparse failure rate"
+      );
+    } finally {
+      consumer.close();
+    }
+  }
+}
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/RebalanceAtScaleIntegrationTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/RebalanceAtScaleIntegrationTest.java
new file mode 100644
index 00000000..8977c30a
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/RebalanceAtScaleIntegrationTest.java
@@ -0,0 +1,538 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.kafka.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/// Rebalance-at-scale integration tests — end-to-end at-least-once across partition handoffs
+/// against a real broker, covering paths the eager-rebalance chaos test does not.
+///
+/// `ChaosRebalanceIntegrationTest` already proves no-loss + no-commit-ahead under the *eager*
+/// (stop-the-world) rebalance protocol. These tests extend that coverage to:
+///
+///   1. **Cooperative-sticky (incremental) rebalance.** With
+///      `partition.assignment.strategy=CooperativeStickyAssignor`, a joining member only gets a
+///      *subset* of partitions revoked from the incumbent rather than a full stop-the-world
+///      revoke-everything cycle. The incumbent keeps draining its retained partitions across the
+///      handoff. This asserts the same no-loss + no-commit-ahead contract holds when the revoke is
+///      partial rather than total.
+///   2. **Mid-flight revocation.** A slow sink keeps records actively in-flight while the
+///      rebalance hits, so the incumbent revokes partitions whose tail is still processing. The
+///      committed prefix must be correct (no commit-ahead) and the new owner re-fetches the rest.
+///   3. **Single-writer command-queue invariant survives a real rebalance.** `onPartitionsRevoked`
+///      must run on the consumer's poll thread (the thread that owns `commandQueue` mutations). An
+///      `assert` inside `KafkaOffsetManager` would AssertionError if it ran off the poll thread
+///      under `-ea`; this test additionally wraps the incumbent's OffsetManager so each revoke
+///      callback records the thread it ran on, then asserts that thread is the named Kafka poll
+///      thread — proving the invariant held under a genuine broker-driven rebalance, not a
+///      synthetic `MockConsumer` rebalance.
+///
+/// All three assert the at-least-once contract with the same `<=` (no-commit-ahead) bound and
+/// exact no-loss set coverage as the chaos test — deliberately NOT exact-once. A partition with
+/// no commit at all is an acceptable uncommitted tail (reprocessed on restart); only partitions
+/// that did commit are bound-checked.
+///
+/// CI-RUN-REQUIRED: these are Testcontainers tests (need a Docker daemon to start the broker).
+/// They compile locally but cannot run where Docker is unavailable; they run in CI.
+@Testcontainers(disabledWithoutDocker = true)
+class RebalanceAtScaleIntegrationTest {
+
+  private static final String KAFKA_VERSION = System.getProperty("kafkaVersion", "4.3.0");
+
+  private static final int PARTITIONS = 6;
+  private static final int RECORD_COUNT = 600;
+
+  @Container
+  static KafkaContainer kafka = new KafkaContainer(
+    DockerImageName.parse("soldevelo/kafka:%s".formatted(KAFKA_VERSION)).asCompatibleSubstituteFor("apache/kafka")
+  );
+
+  /// Cooperative-sticky (incremental) rebalance: a second member joins mid-stream and Kafka
+  /// revokes only a *subset* of the incumbent's partitions instead of a stop-the-world
+  /// revoke-everything cycle. The incumbent keeps draining its retained partitions throughout.
+  /// The no-loss + no-commit-ahead contract must hold across the incremental handoff exactly as
+  /// it does under eager rebalance.
+  @Test
+  void atLeastOnceSurvivesCooperativeStickyRebalance() throws Exception {
+    final var topic = "coop-input-" + System.nanoTime();
+    final var groupId = "coop-group-" + System.nanoTime();
+
+    createTopic(topic);
+    final var producedValues = produceRecords(topic);
+
+    final var observed = ConcurrentHashMap.<String>newKeySet();
+    final var observedTotal = new AtomicInteger(0);
+
+    final var consumerA = buildConsumer(
+      topic,
+      cooperativeStickyProperties(groupId),
+      observed,
+      observedTotal,
+      Duration.ZERO
+    );
+    final var threadA = Thread.ofVirtual().name("coop-consumer-A").start(consumerA::start);
+
+    // Let A claim all partitions and process a slice before the second member joins.
+    awaitObservedAtLeast(observed, RECORD_COUNT / 10, Duration.ofSeconds(20));
+
+    // Second member joins the same group with the same cooperative-sticky strategy. Kafka moves a
+    // subset of A's partitions to B incrementally.
+    final var consumerB = buildConsumer(
+      topic,
+      cooperativeStickyProperties(groupId),
+      observed,
+      observedTotal,
+      Duration.ZERO
+    );
+    final var threadB = Thread.ofVirtual().name("coop-consumer-B").start(consumerB::start);
+
+    try {
+      final var allObserved = awaitObservedAtLeast(observed, producedValues.size(), Duration.ofSeconds(60));
+      assertTrue(
+        allObserved,
+        "Every produced record must be observed at least once after the cooperative-sticky rebalance; saw %d of %d (total deliveries %d)".formatted(
+          observed.size(),
+          producedValues.size(),
+          observedTotal.get()
+        )
+      );
+      assertEquals(producedValues, observed, "Observed set must exactly cover the produced set (no loss).");
+    } finally {
+      consumerA.close();
+      consumerB.close();
+      threadA.join(5000);
+      threadB.join(5000);
+    }
+
+    assertNoCommitAhead(topic, groupId);
+  }
+
+  /// Mid-flight revocation: a slow sink keeps records actively in-flight while the rebalance hits,
+  /// so the incumbent revokes partitions whose tail is still being processed. The committed prefix
+  /// for each revoked partition must reflect only what genuinely finished (no commit-ahead), and
+  /// the new owner re-fetches the unprocessed remainder (no loss). With per-record processing
+  /// trailing the fetch, there is a real in-flight set at the moment of revoke, not a drained log.
+  @Test
+  void midFlightRevocationCommitsPrefixAndReassignsRest() throws Exception {
+    final var topic = "midflight-input-" + System.nanoTime();
+    final var groupId = "midflight-group-" + System.nanoTime();
+
+    createTopic(topic);
+    final var producedValues = produceRecords(topic);
+
+    final var observed = ConcurrentHashMap.<String>newKeySet();
+    final var observedTotal = new AtomicInteger(0);
+
+    // A slow sink (2ms/record) keeps a tail of records in-flight when the rebalance lands, so the
+    // revoke catches partitions mid-processing rather than fully drained.
+    final var sinkDelay = Duration.ofMillis(2);
+
+    final var consumerA = buildConsumer(topic, consumerProperties(groupId), observed, observedTotal, sinkDelay);
+    final var threadA = Thread.ofVirtual().name("midflight-consumer-A").start(consumerA::start);
+
+    awaitObservedAtLeast(observed, RECORD_COUNT / 10, Duration.ofSeconds(30));
+
+    // Second member joins while A is still actively processing in-flight records.
+    final var consumerB = buildConsumer(topic, consumerProperties(groupId), observed, observedTotal, sinkDelay);
+    final var threadB = Thread.ofVirtual().name("midflight-consumer-B").start(consumerB::start);
+
+    try {
+      final var allObserved = awaitObservedAtLeast(observed, producedValues.size(), Duration.ofSeconds(90));
+      assertTrue(
+        allObserved,
+        "Every produced record must be observed at least once across a mid-flight revocation; saw %d of %d (total deliveries %d)".formatted(
+          observed.size(),
+          producedValues.size(),
+          observedTotal.get()
+        )
+      );
+      assertEquals(producedValues, observed, "Observed set must exactly cover the produced set (no loss).");
+    } finally {
+      consumerA.close();
+      consumerB.close();
+      threadA.join(5000);
+      threadB.join(5000);
+    }
+
+    assertNoCommitAhead(topic, groupId);
+  }
+
+  /// Single-writer command-queue invariant under a real rebalance. `onPartitionsRevoked` mutates
+  /// `commandQueue` (draining commands for revoked partitions); that drain is only safe because it
+  /// runs on the consumer's poll thread, the sole writer of the queue. The runtime guard is an
+  /// `assert` inside `KafkaOffsetManager` that would AssertionError if the callback ever fired
+  /// off-thread under `-ea`. This test forces a genuine broker-driven revoke and asserts the
+  /// revoke log record's source thread is the named Kafka poll thread — proving the callback ran
+  /// on the poll thread and the drain stayed correct (data drained, no loss / no commit-ahead).
+  @Test
+  void revokeRunsOnConsumerThreadUnderRealRebalance() throws Exception {
+    final var topic = "single-writer-input-" + System.nanoTime();
+    final var groupId = "single-writer-group-" + System.nanoTime();
+
+    createTopic(topic);
+    final var producedValues = produceRecords(topic);
+
+    // Capture the revoke thread directly at the callback by wrapping the incumbent's OffsetManager
+    // rebalance listener. onPartitionsRevoked fires inside the poll call, so its thread must be the
+    // named Kafka poll thread (the single writer of the command queue). Capturing at the callback
+    // is deterministic: no log-message match, no System.Logger to JUL bridge, no thread-id lookup.
+    final var revokeThreads = new CopyOnWriteArrayList<String>();
+
+    final var observed = ConcurrentHashMap.<String>newKeySet();
+    final var observedTotal = new AtomicInteger(0);
+
+    final var consumerA = buildConsumer(
+      topic,
+      consumerProperties(groupId),
+      observed,
+      observedTotal,
+      Duration.ofMillis(1),
+      revokeThreads
+    );
+    final var threadA = Thread.ofVirtual().name("single-writer-consumer-A").start(consumerA::start);
+
+    awaitObservedAtLeast(observed, RECORD_COUNT / 10, Duration.ofSeconds(30));
+
+    // A second member joins → A's poll loop fires onPartitionsRevoked for the moved partitions.
+    final var consumerB = buildConsumer(
+      topic,
+      consumerProperties(groupId),
+      observed,
+      observedTotal,
+      Duration.ofMillis(1)
+    );
+    final var threadB = Thread.ofVirtual().name("single-writer-consumer-B").start(consumerB::start);
+
+    try {
+      // Wait for a revoke to actually fire on the incumbent.
+      final var sawRevoke = awaitCondition(() -> !revokeThreads.isEmpty(), Duration.ofSeconds(60));
+      assertTrue(sawRevoke, "expected a partition revoke to fire on the incumbent under a real rebalance");
+
+      // Every revoke callback must have run on the named Kafka poll thread — the single writer of
+      // commandQueue. Anything else would mean the listener ran off-thread, which the production
+      // single-writer assert would also have caught by AssertionError.
+      for (final var threadName : revokeThreads) {
+        assertTrue(
+          threadName.startsWith("kafka-consumer-"),
+          "onPartitionsRevoked must run on the Kafka poll thread (single-writer invariant); ran on '%s'".formatted(
+            threadName
+          )
+        );
+      }
+
+      // The drain must stay correct end-to-end: every record still lands at least once.
+      final var allObserved = awaitObservedAtLeast(observed, producedValues.size(), Duration.ofSeconds(60));
+      assertTrue(
+        allObserved,
+        "Every produced record must be observed at least once after the revoke; saw %d of %d (total deliveries %d)".formatted(
+          observed.size(),
+          producedValues.size(),
+          observedTotal.get()
+        )
+      );
+      assertEquals(producedValues, observed, "Observed set must exactly cover the produced set (no loss).");
+    } finally {
+      consumerA.close();
+      consumerB.close();
+      threadA.join(5000);
+      threadB.join(5000);
+    }
+
+    assertNoCommitAhead(topic, groupId);
+  }
+
+  /// No commit-ahead / no silent skip: where a partition has a committed offset, it must never
+  /// exceed the log end. The bound is `<=`, NOT `==`, and a partition may legitimately have NO
+  /// committed offset at all — at-least-once does not promise the commit reaches log-end at an
+  /// arbitrary shutdown across a rebalance. A fully-processed-but-uncommitted partition is simply
+  /// re-read and reprocessed on restart (at-least-once tolerance, not loss). Combined with the
+  /// no-loss assertion at each call site, this is the at-least-once contract. Requiring a commit
+  /// on every partition, or `== logEnd`, would assert an exactly-once / complete-drain property
+  /// kpipe deliberately does not claim.
+  private void assertNoCommitAhead(final String topic, final String groupId) throws Exception {
+    final var endOffsets = endOffsets(topic);
+    final var committed = committedOffsets(groupId);
+
+    for (final var entry : endOffsets.entrySet()) {
+      final var tp = entry.getKey();
+      final var logEnd = entry.getValue();
+      final var committedMeta = committed.get(tp);
+      if (committedMeta == null) {
+        continue;
+      }
+      assertTrue(
+        committedMeta.offset() <= logEnd,
+        "Committed offset for %s must not exceed the log end (no commit-ahead). committed=%d logEnd=%d".formatted(
+          tp,
+          committedMeta.offset(),
+          logEnd
+        )
+      );
+    }
+  }
+
+  private void createTopic(final String topic) throws Exception {
+    try (final var admin = Admin.create(adminProperties())) {
+      admin.createTopics(List.of(new NewTopic(topic, PARTITIONS, (short) 1))).all().get(30, TimeUnit.SECONDS);
+    }
+  }
+
+  /// Produces [#RECORD_COUNT] keyed records spread across all partitions. Keys are
+  /// `key-<i % PARTITIONS>` so the default partitioner fans them out, and values are unique
+  /// (`val-<i>`) so the sink can assert exact set coverage. Returns the set of produced values.
+  private Set<String> produceRecords(final String topic) throws Exception {
+    final var values = ConcurrentHashMap.<String>newKeySet();
+    final var props = producerProperties();
+    try (final var producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) {
+      final var sends = new ArrayList<Future<?>>();
+      for (int i = 0; i < RECORD_COUNT; i++) {
+        final var value = "val-" + i;
+        values.add(value);
+        final var key = ("key-" + (i % PARTITIONS)).getBytes();
+        sends.add(producer.send(new ProducerRecord<>(topic, key, value.getBytes())));
+      }
+      producer.flush();
+      for (final var send : sends) {
+        send.get(30, TimeUnit.SECONDS);
+      }
+    }
+    return Set.copyOf(values);
+  }
+
+  /// Builds a PARALLEL consumer over `topic`. A non-zero `sinkDelay` slows each record so the
+  /// processing trails the fetch, keeping a tail of records in-flight when a rebalance lands.
+  private KPipeConsumer<byte[]> buildConsumer(
+    final String topic,
+    final Properties props,
+    final Set<String> observed,
+    final AtomicInteger observedTotal,
+    final Duration sinkDelay
+  ) {
+    return buildConsumer(topic, props, observed, observedTotal, sinkDelay, null);
+  }
+
+  /// Builds a PARALLEL consumer over `topic`. When `revokeThreads` is non-null, the consumer's
+  /// OffsetManager is wrapped so each `onPartitionsRevoked` callback records the thread it ran on —
+  /// the deterministic single-writer probe used by `revokeRunsOnConsumerThreadUnderRealRebalance`.
+  private KPipeConsumer<byte[]> buildConsumer(
+    final String topic,
+    final Properties props,
+    final Set<String> observed,
+    final AtomicInteger observedTotal,
+    final Duration sinkDelay,
+    final List<String> revokeThreads
+  ) {
+    final var builder = KPipeConsumer.<byte[]>builder()
+      .withProperties(props)
+      .withTopic(topic)
+      .withProcessingMode(ProcessingMode.PARALLEL)
+      .withPipeline(
+        TestPipelines.sideEffect(value -> {
+          if (!sinkDelay.isZero()) {
+            try {
+              TimeUnit.NANOSECONDS.sleep(sinkDelay.toNanos());
+            } catch (final InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+          }
+          observed.add(new String(value));
+          observedTotal.incrementAndGet();
+          return value;
+        })
+      );
+    if (revokeThreads != null) {
+      builder.withOffsetManagerProvider(consumer ->
+        new RevokeCapturingOffsetManager(
+          KafkaOffsetManager.<byte[]>builder(consumer).withCommandQueue(builder.getCommandQueue()).build(),
+          revokeThreads
+        )
+      );
+    }
+    return builder.build();
+  }
+
+  /// Decorator that records the calling thread of every non-empty `onPartitionsRevoked` into
+  /// `revokeThreads`, then delegates. Everything else passes straight through to the real manager.
+  /// Capturing at the callback is deterministic — no log-string match or thread-id resolution.
+  private static final class RevokeCapturingOffsetManager implements OffsetManager<byte[]> {
+
+    private final OffsetManager<byte[]> delegate;
+    private final List<String> revokeThreads;
+
+    RevokeCapturingOffsetManager(final OffsetManager<byte[]> delegate, final List<String> revokeThreads) {
+      this.delegate = delegate;
+      this.revokeThreads = revokeThreads;
+    }
+
+    @Override
+    public ConsumerRebalanceListener createRebalanceListener() {
+      final var inner = delegate.createRebalanceListener();
+      return new ConsumerRebalanceListener() {
+        @Override
+        public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
+          if (!partitions.isEmpty()) revokeThreads.add(Thread.currentThread().getName());
+          inner.onPartitionsRevoked(partitions);
+        }
+
+        @Override
+        public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
+          inner.onPartitionsAssigned(partitions);
+        }
+      };
+    }
+
+    @Override
+    public OffsetManager<byte[]> start() {
+      delegate.start();
+      return this;
+    }
+
+    @Override
+    public OffsetManager<byte[]> stop() {
+      delegate.stop();
+      return this;
+    }
+
+    @Override
+    public void trackOffset(final ConsumerRecord<byte[], byte[]> record) {
+      delegate.trackOffset(record);
+    }
+
+    @Override
+    public void markOffsetProcessed(final ConsumerRecord<byte[], byte[]> record) {
+      delegate.markOffsetProcessed(record);
+    }
+
+    @Override
+    public void notifyCommitComplete(final String commitId, final boolean success) {
+      delegate.notifyCommitComplete(commitId, success);
+    }
+
+    @Override
+    public OffsetState getState() {
+      return delegate.getState();
+    }
+
+    @Override
+    public boolean isRunning() {
+      return delegate.isRunning();
+    }
+
+    @Override
+    public Map<String, Object> getStatistics() {
+      return delegate.getStatistics();
+    }
+
+    @Override
+    public void close() {
+      delegate.close();
+    }
+  }
+
+  /// Polls until the observed set reaches `target` distinct values or the deadline passes.
+  /// Returns true if the target was reached.
+  private boolean awaitObservedAtLeast(final Set<String> observed, final int target, final Duration timeout)
+    throws InterruptedException {
+    return awaitCondition(() -> observed.size() >= target, timeout);
+  }
+
+  /// Polls a boolean condition until it holds or the deadline passes. Returns the final state.
+  private boolean awaitCondition(final java.util.function.BooleanSupplier condition, final Duration timeout)
+    throws InterruptedException {
+    final var deadline = System.currentTimeMillis() + timeout.toMillis();
+    while (System.currentTimeMillis() < deadline) {
+      if (condition.getAsBoolean()) {
+        return true;
+      }
+      TimeUnit.MILLISECONDS.sleep(100);
+    }
+    return condition.getAsBoolean();
+  }
+
+  private Map<TopicPartition, Long> endOffsets(final String topic) {
+    try (
+      final var consumer = new KafkaConsumer<>(
+        consumerProperties("end-offset-probe-" + System.nanoTime()),
+        new ByteArrayDeserializer(),
+        new ByteArrayDeserializer()
+      )
+    ) {
+      final var partitions = new ArrayList<TopicPartition>();
+      for (int p = 0; p < PARTITIONS; p++) {
+        partitions.add(new TopicPartition(topic, p));
+      }
+      return consumer.endOffsets(partitions, Duration.ofSeconds(15));
+    }
+  }
+
+  private Map<TopicPartition, OffsetAndMetadata> committedOffsets(final String groupId) throws Exception {
+    try (final var admin = Admin.create(adminProperties())) {
+      return admin.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get(15, TimeUnit.SECONDS);
+    }
+  }
+
+  private Properties adminProperties() {
+    final var props = new Properties();
+    props.put("bootstrap.servers", kafka.getBootstrapServers());
+    return props;
+  }
+
+  private Properties producerProperties() {
+    final var props = new Properties();
+    props.put("bootstrap.servers", kafka.getBootstrapServers());
+    props.put("acks", "all");
+    return props;
+  }
+
+  /// Consumer properties with the CooperativeStickyAssignor → incremental rebalance: the
+  /// incumbent keeps its retained partitions and only the moved subset is revoked, distinct from
+  /// the eager chaos-test path. A fresh instance per call so two members never share Properties.
+  private Properties cooperativeStickyProperties(final String groupId) {
+    final var props = consumerProperties(groupId);
+    props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getName());
+    return props;
+  }
+
+  private Properties consumerProperties(final String groupId) {
+    final var props = new Properties();
+    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+    // Short session timeout so a joining member triggers the rebalance promptly within the window.
+    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
+    props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
+    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+    return props;
+  }
+}
diff --git a/lib/kpipe-core/src/test/java/io/github/eschizoid/kpipe/registry/TypedPipelineSkipBytesTest.java b/lib/kpipe-core/src/test/java/io/github/eschizoid/kpipe/registry/TypedPipelineSkipBytesTest.java
new file mode 100644
index 00000000..b73e47c2
--- /dev/null
+++ b/lib/kpipe-core/src/test/java/io/github/eschizoid/kpipe/registry/TypedPipelineSkipBytesTest.java
@@ -0,0 +1,143 @@
+package io.github.eschizoid.kpipe.registry;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Arrays;
+import org.junit.jupiter.api.Test;
+
+/// Pins the wire-prefix stripping done by the pipeline built from [TypedPipelineBuilder].
+///
+/// `skipBytes(n)` is the seam that strips Confluent magic-byte envelopes (5 bytes for Avro,
+/// 6 for single-message Protobuf) before the format ever sees the payload. An off-by-one here
+/// is the classic silent-corruption bug: skip one byte too few and the format decodes the
+/// stray prefix byte as payload; skip one too many and the first real byte is lost. These
+/// tests verify the slice handed to the format is exactly `data[n..]`, and that a payload no
+/// longer than the prefix surfaces through the no-null contract rather than slipping through
+/// as a quietly filtered record.
+class TypedPipelineSkipBytesTest {
+
+  /// Records the exact bytes the format was asked to deserialize so the test can assert on the
+  /// post-skip slice. Echoes the bytes back as the "decoded" value.
+  private static final class CapturingFormat implements MessageFormat<byte[]> {
+
+    byte[] lastSeen;
+
+    @Override
+    public byte[] serialize(final byte[] data) {
+      return data;
+    }
+
+    @Override
+    public byte[] deserialize(final byte[] data) {
+      lastSeen = data;
+      return data;
+    }
+  }
+
+  private static MessagePipeline<byte[]> pipelineWithSkip(final MessageFormat<byte[]> format, final int skip) {
+    return new TypedPipelineBuilder<>(format, new MessageProcessorRegistry()).skipBytes(skip).build();
+  }
+
+  @Test
+  void stripsExactlyTheConfiguredPrefixForAvroEnvelope() {
+    // 1-byte magic + 4-byte schema id + 3-byte payload.
+    final var record = new byte[] { 0x00, 0x00, 0x00, 0x00, 0x42, 0x11, 0x22, 0x33 };
+    final var format = new CapturingFormat();
+
+    final var decoded = pipelineWithSkip(format, 5).deserialize(record);
+
+    final var expectedPayload = new byte[] { 0x11, 0x22, 0x33 };
+    assertArrayEquals(expectedPayload, format.lastSeen, "format must see only the bytes after the 5-byte envelope");
+    assertArrayEquals(expectedPayload, decoded, "the decoded value is the post-skip slice");
+  }
+
+  @Test
+  void stripsExactlyTheConfiguredPrefixForProtobufEnvelope() {
+    // 1-byte magic + 4-byte schema id + 1-byte message-index header + 2-byte payload.
+    final var record = new byte[] { 0x00, 0x00, 0x00, 0x00, 0x07, 0x00, (byte) 0xAB, (byte) 0xCD };
+    final var format = new CapturingFormat();
+
+    pipelineWithSkip(format, 6).deserialize(record);
+
+    assertArrayEquals(
+      new byte[] { (byte) 0xAB, (byte) 0xCD },
+      format.lastSeen,
+      "format must see only the bytes after the 6-byte envelope"
+    );
+  }
+
+  @Test
+  void offByOneSkipLeaksOrDropsAByteRatherThanMatching() {
+    // Same Avro-shaped record; only skip(5) yields the clean payload. skip(4) leaks the last
+    // envelope byte; skip(6) eats the first payload byte. Both are silent corruption, so pin
+    // the boundary so a regression that changes the slice length is caught.
+    final var record = new byte[] { 0x00, 0x00, 0x00, 0x00, 0x42, 0x11, 0x22, 0x33 };
+    final var clean = new byte[] { 0x11, 0x22, 0x33 };
+
+    final var four = new CapturingFormat();
+    pipelineWithSkip(four, 4).deserialize(record);
+    assertArrayEquals(new byte[] { 0x42, 0x11, 0x22, 0x33 }, four.lastSeen, "skip(4) leaks the trailing envelope byte");
+    assertNotEquals(Arrays.toString(clean), Arrays.toString(four.lastSeen));
+
+    final var six = new CapturingFormat();
+    pipelineWithSkip(six, 6).deserialize(record);
+    assertArrayEquals(new byte[] { 0x22, 0x33 }, six.lastSeen, "skip(6) drops the first payload byte");
+    assertNotEquals(Arrays.toString(clean), Arrays.toString(six.lastSeen));
+  }
+
+  @Test
+  void payloadShorterThanPrefixDeserializesToNull() {
+    final var format = new CapturingFormat();
+    // 4 bytes of envelope, none of payload, against a 5-byte skip.
+    final var tooShort = new byte[] { 0x00, 0x00, 0x00, 0x00 };
+
+    assertNull(pipelineWithSkip(format, 5).deserialize(tooShort), "no payload after the prefix decodes to null");
+    assertNull(format.lastSeen, "the format must never be invoked when there is nothing left to decode");
+  }
+
+  @Test
+  void payloadExactlyEqualToPrefixLengthDeserializesToNull() {
+    final var format = new CapturingFormat();
+    // Exactly the 5-byte envelope, zero payload. The boundary is `data.length <= skip`.
+    final var envelopeOnly = new byte[] { 0x00, 0x00, 0x00, 0x00, 0x42 };
+
+    assertNull(pipelineWithSkip(format, 5).deserialize(envelopeOnly), "an envelope with no payload decodes to null");
+    assertNull(format.lastSeen);
+  }
+
+  @Test
+  void deserializeOrFailRejectsTheTruncatedPrefixAsAContractViolation() {
+    // The honest end-to-end path: a too-short record must not be reported as a silently
+    // filtered record. deserializeOrFail turns the null into the retryable no-null violation.
+    final var pipeline = pipelineWithSkip(new CapturingFormat(), 5);
+    final var tooShort = new byte[] { 0x00, 0x00, 0x00 };
+
+    final var ex = assertThrows(IllegalStateException.class, () -> pipeline.deserializeOrFail(tooShort));
+    assertEquals("deserialize() returned null — implementations must throw on malformed input", ex.getMessage());
+  }
+
+  @Test
+  void singleTrailingByteAfterPrefixIsKept() {
+    final var format = new CapturingFormat();
+    // One byte more than the prefix — the smallest non-null slice.
+    final var record = new byte[] { 0x00, 0x00, 0x00, 0x00, 0x42, 0x7e };
+
+    pipelineWithSkip(format, 5).deserialize(record);
+
+    assertArrayEquals(new byte[] { 0x7e }, format.lastSeen);
+  }
+
+  @Test
+  void zeroSkipHandsTheWholeRecordToTheFormat() {
+    final var format = new CapturingFormat();
+    final var record = new byte[] { 0x11, 0x22, 0x33 };
+
+    pipelineWithSkip(format, 0).deserialize(record);
+
+    assertArrayEquals(record, format.lastSeen, "skip(0) is a passthrough — no slicing");
+  }
+}
diff --git a/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatEnvelopeEdgeCasesTest.java b/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatEnvelopeEdgeCasesTest.java
new file mode 100644
index 00000000..3dda2ca8
--- /dev/null
+++ b/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatEnvelopeEdgeCasesTest.java
@@ -0,0 +1,167 @@
+package io.github.eschizoid.kpipe.format.avro;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.registry.MessagePipeline;
+import io.github.eschizoid.kpipe.registry.MessageProcessorRegistry;
+import io.github.eschizoid.kpipe.registry.Result;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.EncoderFactory;
+import org.junit.jupiter.api.Test;
+
+/// Avro edge cases that live on the static-mode + `skipBytes` path: the exact 5-byte Confluent
+/// envelope boundary, off-by-one wire-prefix stripping, and oversized payloads. These exercise
+/// the real pipeline built from the registry, so the slice fed to the codec is whatever the
+/// builder produced — the surface where an envelope off-by-one corrupts silently.
+class AvroFormatEnvelopeEdgeCasesTest {
+
+  private static final String SCHEMA_JSON = """
+    {
+      "type": "record",
+      "name": "Simple",
+      "fields": [
+        {"name": "value", "type": "string"}
+      ]
+    }""";
+
+  private static final MessageProcessorRegistry REGISTRY = new MessageProcessorRegistry();
+
+  /// Drives `bytes` through deserialize -> process -> serialize, re-throwing a failure cause.
+  private static <T> byte[] roundTrip(final MessagePipeline<T> pipeline, final byte[] bytes) {
+    final var deserialized = pipeline.deserializeOrFail(bytes);
+    return switch (pipeline.process(deserialized)) {
+      case Result.Passed<T> p -> pipeline.serialize(p.value());
+      case Result.Filtered<T> _ -> null;
+      case Result.Failed<T> f -> {
+        if (f.cause() instanceof RuntimeException re) throw re;
+        if (f.cause() instanceof Error err) throw err;
+        throw new RuntimeException(f.cause());
+      }
+    };
+  }
+
+  /// Encodes a record's raw Avro binary (no envelope).
+  private static byte[] encode(final Schema schema, final String value) throws IOException {
+    final var record = new GenericData.Record(schema);
+    record.put("value", value);
+    try (final var out = new ByteArrayOutputStream()) {
+      final var writer = new GenericDatumWriter<GenericRecord>(schema);
+      final var encoder = EncoderFactory.get().binaryEncoder(out, null);
+      writer.write(record, encoder);
+      encoder.flush();
+      return out.toByteArray();
+    }
+  }
+
+  /// Prepends the 5-byte Confluent envelope (1-byte magic 0x00 + 4-byte big-endian schema id).
+  private static byte[] withEnvelope(final int schemaId, final byte[] payload) {
+    final var framed = new byte[5 + payload.length];
+    framed[0] = 0x00;
+    ByteBuffer.wrap(framed, 1, 4).putInt(schemaId);
+    System.arraycopy(payload, 0, framed, 5, payload.length);
+    return framed;
+  }
+
+  @Test
+  void skipBytesFiveStripsTheEnvelopeAndDecodesCleanly() throws IOException {
+    final var format = AvroFormat.of(SCHEMA_JSON);
+    final var payload = encode(format.schema(), "hello");
+    final var framed = withEnvelope(42, payload);
+
+    final var pipeline = REGISTRY.pipeline(format).skipBytes(5).build();
+    final var decoded = pipeline.deserializeOrFail(framed);
+
+    assertNotNull(decoded);
+    assertEquals("hello", decoded.get("value").toString());
+  }
+
+  @Test
+  void offByOneSkipFourCorruptsOrFails() throws IOException {
+    // Leaving the trailing schema-id byte in front of the payload either makes Avro throw or
+    // decode a different string — never the clean value. Pins that exactly 5 is required.
+    final var format = AvroFormat.of(SCHEMA_JSON);
+    final var payload = encode(format.schema(), "hello");
+    final var framed = withEnvelope(42, payload);
+
+    final var pipeline = REGISTRY.pipeline(format).skipBytes(4).build();
+
+    var corrupted = false;
+    try {
+      final var decoded = pipeline.deserializeOrFail(framed);
+      corrupted = decoded == null || !"hello".equals(decoded.get("value").toString());
+    } catch (final RuntimeException e) {
+      // A throw is an acceptable "did not silently match" outcome — but require it to be the
+      // format's decode failure, not an unrelated NPE/wiring error that would green vacuously.
+      assertDecodeFailure(e);
+      corrupted = true;
+    }
+    assertTrue(corrupted, "skip(4) must not decode to the clean value");
+  }
+
+  @Test
+  void offByOneSkipSixCorruptsOrFails() throws IOException {
+    // Eating the first payload byte shifts every downstream field into corruption.
+    final var format = AvroFormat.of(SCHEMA_JSON);
+    final var payload = encode(format.schema(), "hello");
+    final var framed = withEnvelope(42, payload);
+
+    final var pipeline = REGISTRY.pipeline(format).skipBytes(6).build();
+
+    var corrupted = false;
+    try {
+      final var decoded = pipeline.deserializeOrFail(framed);
+      corrupted = decoded == null || !"hello".equals(decoded.get("value").toString());
+    } catch (final RuntimeException e) {
+      assertDecodeFailure(e);
+      corrupted = true;
+    }
+    assertTrue(corrupted, "skip(6) must not decode to the clean value");
+  }
+
+  @Test
+  void oversizedPayloadRoundTripsWithoutTruncation() throws IOException {
+    // A multi-megabyte string value exercises the codec's buffer growth on a large record.
+    final var format = AvroFormat.of(SCHEMA_JSON);
+    final var big = "x".repeat(2_000_000);
+    final var bytes = encode(format.schema(), big);
+
+    final var pipeline = REGISTRY.pipeline(format).build();
+    final var result = roundTrip(pipeline, bytes);
+
+    assertNotNull(result);
+    assertArrayEquals(bytes, result, "an oversized payload must round-trip byte-for-byte");
+    assertEquals(big, format.deserialize(result).get("value").toString());
+  }
+
+  @Test
+  void registryModeDecodesValidEnvelopeWithoutSkipBytes() throws IOException {
+    // The complement to skipBytes-5: registry mode consumes the envelope itself, so the raw
+    // framed record decodes straight through with no skip configured.
+    final var schema = new Schema.Parser().parse(SCHEMA_JSON);
+    final var payload = encode(schema, "via-registry");
+    final var framed = withEnvelope(7, payload);
+
+    final var format = AvroFormat.withRegistry(id -> id == 7 ? SCHEMA_JSON : null);
+    final var decoded = format.deserialize(framed);
+
+    assertEquals("via-registry", decoded.get("value").toString());
+  }
+
+  /// Confirms a caught throw is the Avro codec's own decode failure — not an unrelated
+  /// NPE/wiring error — so an off-by-one test cannot pass on the wrong exception.
+  private static void assertDecodeFailure(final RuntimeException e) {
+    assertTrue(
+      e.getMessage() != null && e.getMessage().contains("AvroFormat.deserialize failed"),
+      () -> "expected an Avro decode failure, got: " + e
+    );
+  }
+}
diff --git a/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatSchemaEvolutionTest.java b/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatSchemaEvolutionTest.java
new file mode 100644
index 00000000..ac48c706
--- /dev/null
+++ b/lib/kpipe-format-avro/src/test/java/io/github/eschizoid/kpipe/format/avro/AvroFormatSchemaEvolutionTest.java
@@ -0,0 +1,325 @@
+package io.github.eschizoid.kpipe.format.avro;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.registry.SchemaResolver;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.EncoderFactory;
+import org.junit.jupiter.api.Test;
+
+/// End-to-end schema-evolution behaviour for [AvroFormat].
+///
+/// The point of registry-backed decoding is that every record is decoded against the *exact*
+/// writer schema that produced it — resolved per record from the wire envelope — rather than a
+/// single schema chosen once at startup. That is what keeps a consumer correct across producer
+/// schema rolls. These tests prove three things directly:
+///
+/// 1. A registry-backed format decodes records written under different schema versions (added
+///    field, removed field) correctly, because the writer schema is looked up per record.
+/// 2. A static-mode format bound to an *old* schema silently mis-decodes bytes written under a
+///    newer, incompatible schema — the corruption the registry path exists to prevent. This is
+///    the contrast that justifies per-record lookup even when the registry enforces compatible
+///    evolution.
+/// 3. Stripping the Confluent envelope before a registry-backed format sees it (the
+///    `withSchemaRegistry(...)` + `skipBytes(5)` misconfiguration) surfaces a decode error rather
+///    than quietly returning wrong data.
+class AvroFormatSchemaEvolutionTest {
+
+  /// v1: id + name.
+  private static final String USER_V1 = """
+    {"type":"record","name":"User","fields":[
+      {"name":"id","type":"string"},
+      {"name":"name","type":"string"}
+    ]}""";
+
+  /// v2: appends an optional `email` (union with null default) — BACKWARD/FORWARD compatible.
+  private static final String USER_V2 = """
+    {"type":"record","name":"User","fields":[
+      {"name":"id","type":"string"},
+      {"name":"name","type":"string"},
+      {"name":"email","type":["null","string"],"default":null}
+    ]}""";
+
+  /// v3: drops `name`, keeps `id` + optional `email`. A field removal — the kind of non-append
+  /// evolution a static v1-bound reader cannot decode safely.
+  private static final String USER_V3 = """
+    {"type":"record","name":"User","fields":[
+      {"name":"id","type":"string"},
+      {"name":"email","type":["null","string"],"default":null}
+    ]}""";
+
+  /// Two records with a leading int field, used to show concretely what goes wrong when a reader
+  /// decodes payload framed by a different writer schema: the field a static reader lands on no
+  /// longer matches the bytes the writer laid down.
+  private static final String EVENT_WIDE = """
+    {"type":"record","name":"Event","fields":[
+      {"name":"seq","type":"long"},
+      {"name":"flag","type":"boolean"},
+      {"name":"label","type":"string"}
+    ]}""";
+
+  private static final String EVENT_NARROW = """
+    {"type":"record","name":"Event","fields":[
+      {"name":"seq","type":"long"},
+      {"name":"label","type":"string"}
+    ]}""";
+
+  /// Encodes `record` under the Confluent wire envelope: 1-byte magic (0x00) + 4-byte big-endian
+  /// schema id + Avro binary payload written with the record's own schema.
+  private static byte[] envelope(final int schemaId, final GenericRecord record) throws IOException {
+    final var bytes = new ByteArrayOutputStream();
+    bytes.write(0x00);
+    bytes.write(ByteBuffer.allocate(4).putInt(schemaId).array());
+    final var writer = new GenericDatumWriter<GenericRecord>(record.getSchema());
+    final var encoder = EncoderFactory.get().binaryEncoder(bytes, null);
+    writer.write(record, encoder);
+    encoder.flush();
+    return bytes.toByteArray();
+  }
+
+  /// Raw Avro binary payload (no envelope) for the static-mode corruption contrast.
+  private static byte[] payload(final GenericRecord record) throws IOException {
+    final var bytes = new ByteArrayOutputStream();
+    final var writer = new GenericDatumWriter<GenericRecord>(record.getSchema());
+    final var encoder = EncoderFactory.get().binaryEncoder(bytes, null);
+    writer.write(record, encoder);
+    encoder.flush();
+    return bytes.toByteArray();
+  }
+
+  /// Serves preregistered schemas by id and counts lookups.
+  private static final class FakeResolver implements SchemaResolver {
+
+    final Map<Integer, String> schemas = new HashMap<>();
+    final AtomicInteger calls = new AtomicInteger();
+
+    FakeResolver put(final int id, final String json) {
+      schemas.put(id, json);
+      return this;
+    }
+
+    @Override
+    public String lookupById(final int schemaId) {
+      calls.incrementAndGet();
+      final var json = schemas.get(schemaId);
+      if (json == null) throw new RuntimeException("Unknown schema id " + schemaId);
+      return json;
+    }
+  }
+
+  @Test
+  void registryDecodesAddedFieldRecordAgainstItsOwnWriterSchema() throws IOException {
+    final var v2 = new Schema.Parser().parse(USER_V2);
+    final var newRecord = new GenericData.Record(v2);
+    newRecord.put("id", "u-2");
+    newRecord.put("name", "New");
+    newRecord.put("email", "new@example.com");
+
+    final var resolver = new FakeResolver().put(1, USER_V1).put(2, USER_V2);
+    final var format = AvroFormat.withRegistry(resolver);
+
+    final var decoded = format.deserialize(envelope(2, newRecord));
+
+    // The appended field is present because the v2 writer schema was resolved for this record;
+    // a v1-bound static reader would never surface it.
+    assertEquals("u-2", decoded.get("id").toString());
+    assertEquals("New", decoded.get("name").toString());
+    assertEquals("new@example.com", decoded.get("email").toString());
+  }
+
+  @Test
+  void registryDecodesRemovedFieldRecordAgainstItsOwnWriterSchema() throws IOException {
+    final var v3 = new Schema.Parser().parse(USER_V3);
+    final var record = new GenericData.Record(v3);
+    record.put("id", "u-3");
+    record.put("email", "drop@example.com");
+
+    final var resolver = new FakeResolver().put(1, USER_V1).put(3, USER_V3);
+    final var format = AvroFormat.withRegistry(resolver);
+
+    final var decoded = format.deserialize(envelope(3, record));
+
+    // `name` was removed in v3; decoding against the v3 writer schema yields a record that has no
+    // `name` field at all rather than mis-reading the following bytes as a string.
+    assertEquals("u-3", decoded.get("id").toString());
+    assertEquals("drop@example.com", decoded.get("email").toString());
+    assertNull(
+      decoded.getSchema().getField("name"),
+      "v3 writer schema has no name field, so the decoded record must not either"
+    );
+  }
+
+  @Test
+  void registryDecodesInterleavedSchemaVersionsCorrectly() throws IOException {
+    // A single format instance consuming a stream that interleaves v1, v2 and v3 records — the
+    // realistic during-a-roll case. Each record carries its own schema id and must decode against
+    // its own writer schema, independent of arrival order.
+    final var v1 = new Schema.Parser().parse(USER_V1);
+    final var v2 = new Schema.Parser().parse(USER_V2);
+    final var v3 = new Schema.Parser().parse(USER_V3);
+
+    final var r1 = new GenericData.Record(v1);
+    r1.put("id", "a");
+    r1.put("name", "Alice");
+
+    final var r2 = new GenericData.Record(v2);
+    r2.put("id", "b");
+    r2.put("name", "Bob");
+    r2.put("email", "bob@example.com");
+
+    final var r3 = new GenericData.Record(v3);
+    r3.put("id", "c");
+    r3.put("email", "carol@example.com");
+
+    final var resolver = new FakeResolver().put(1, USER_V1).put(2, USER_V2).put(3, USER_V3);
+    final var format = AvroFormat.withRegistry(resolver);
+
+    final var d2 = format.deserialize(envelope(2, r2));
+    final var d1 = format.deserialize(envelope(1, r1));
+    final var d3 = format.deserialize(envelope(3, r3));
+
+    assertEquals("Alice", d1.get("name").toString());
+    assertNull(d1.getSchema().getField("email"), "v1 record has no email field");
+
+    assertEquals("Bob", d2.get("name").toString());
+    assertEquals("bob@example.com", d2.get("email").toString());
+
+    assertEquals("c", d3.get("id").toString());
+    assertEquals("carol@example.com", d3.get("email").toString());
+    assertNull(d3.getSchema().getField("name"), "v3 record has no name field");
+  }
+
+  @Test
+  void staticReaderBoundToOldSchemaMisDecodesNewerIncompatibleBytes() throws IOException {
+    // The corruption the registry path prevents. EVENT_WIDE writes [seq, flag, label]; a static
+    // reader bound to EVENT_NARROW expects [seq, label]. After `seq` the narrow reader treats the
+    // boolean `flag` byte as the start of `label`'s length-prefixed string — so `label` decodes to
+    // garbage (or the decode blows up), never the value the producer wrote. No exception is
+    // guaranteed: silent corruption is the whole danger.
+    final var wide = new Schema.Parser().parse(EVENT_WIDE);
+    final var record = new GenericData.Record(wide);
+    record.put("seq", 7L);
+    record.put("flag", true);
+    record.put("label", "real-label");
+    final var wideBytes = payload(record);
+
+    final var staticNarrow = new AvroFormat(new Schema.Parser().parse(EVENT_NARROW));
+
+    // Either the mis-framed decode throws, or it returns a record whose `label` is NOT the value
+    // the producer actually wrote. Both outcomes prove a static old-schema reader cannot be
+    // trusted on newer incompatible bytes.
+    String corruptLabel = null;
+    try {
+      final var misdecoded = staticNarrow.deserialize(wideBytes);
+      corruptLabel = String.valueOf(misdecoded.get("label"));
+    } catch (final RuntimeException expected) {
+      // Mis-framing surfaced as a decode error — also an acceptable "did not silently succeed",
+      // but require it to be the codec's own failure so an unrelated error can't green this.
+      assertTrue(
+        expected.getMessage() != null && expected.getMessage().contains("AvroFormat.deserialize failed"),
+        () -> "expected an Avro decode failure, got: " + expected
+      );
+      return;
+    }
+    assertNotEquals(
+      "real-label",
+      corruptLabel,
+      "a static narrow reader must not reproduce the wide writer's label — that would mean the " +
+        "corruption went undetected"
+    );
+  }
+
+  @Test
+  void registryDecodesNewerBytesThatStaticOldReaderWouldCorrupt() throws IOException {
+    // Same wide-vs-narrow framing, but now the wide bytes carry an envelope and the registry
+    // resolves the wide writer schema per record. The label decodes to exactly what the producer
+    // wrote — proving the registry path is correct precisely where the static old reader corrupts.
+    final var wide = new Schema.Parser().parse(EVENT_WIDE);
+    final var record = new GenericData.Record(wide);
+    record.put("seq", 7L);
+    record.put("flag", true);
+    record.put("label", "real-label");
+
+    final var resolver = new FakeResolver().put(9, EVENT_WIDE);
+    final var format = AvroFormat.withRegistry(resolver);
+
+    final var decoded = format.deserialize(envelope(9, record));
+
+    assertEquals(7L, decoded.get("seq"));
+    assertEquals(true, decoded.get("flag"));
+    assertEquals("real-label", decoded.get("label").toString());
+  }
+
+  @Test
+  void skipBytesBeforeRegistryFormatStripsEnvelopeAndForcesDecodeError() throws IOException {
+    // Combining withSchemaRegistry(...) with skipBytes(5) double-strips: the 5-byte envelope is
+    // removed upstream, so the format sees the bare Avro payload where it expects the magic byte +
+    // schema id. Simulate skipBytes(5) by hand-stripping the envelope, then feed it to a
+    // registry-backed format. The first payload byte is no longer 0x00, so the format must reject
+    // it rather than read a bogus schema id and silently decode wrong data.
+    final var v2 = new Schema.Parser().parse(USER_V2);
+    final var record = new GenericData.Record(v2);
+    record.put("id", "u-2");
+    record.put("name", "New");
+    record.put("email", "new@example.com");
+
+    final var framed = envelope(2, record);
+    final var stripped = new byte[framed.length - 5];
+    System.arraycopy(framed, 5, stripped, 0, stripped.length);
+
+    final var resolver = new FakeResolver().put(2, USER_V2);
+    final var format = AvroFormat.withRegistry(resolver);
+
+    final var ex = assertThrows(RuntimeException.class, () -> format.deserialize(stripped));
+    // The Avro payload for this record does not begin with the 0x00 magic byte, so the envelope
+    // check fires before any schema lookup happens.
+    assertTrue(
+      ex.getMessage().contains("magic byte") || ex.getMessage().contains("envelope"),
+      "double-stripped record must fail the envelope check, not decode silently; got: " + ex.getMessage()
+    );
+    assertEquals(0, resolver.calls.get(), "envelope must be rejected before any schema lookup");
+  }
+
+  @Test
+  void skipBytesShorteningEnvelopeBelowFiveBytesIsRejected() throws IOException {
+    // Degenerate variant: the record is short enough that stripping 5 bytes leaves fewer than the
+    // envelope length, so the too-short guard fires instead of the magic-byte guard. Still a hard
+    // error, never silent.
+    final var v1 = new Schema.Parser().parse(USER_V1);
+    final var record = new GenericData.Record(v1);
+    record.put("id", "");
+    record.put("name", "");
+
+    final var framed = envelope(1, record);
+    final var stripped = new byte[Math.max(0, framed.length - 5)];
+    System.arraycopy(framed, 5, stripped, 0, stripped.length);
+
+    final var format = AvroFormat.withRegistry(new FakeResolver().put(1, USER_V1));
+
+    // Empty-string fields encode to two zero-length varints, so the stripped payload is 2 bytes —
+    // below ENVELOPE_LENGTH. deserialize returns null only for null/empty input; a 2-byte input
+    // reaches the too-short envelope guard.
+    if (stripped.length == 0) {
+      assertNull(format.deserialize(stripped));
+    } else {
+      final var ex = assertThrows(RuntimeException.class, () -> format.deserialize(stripped));
+      assertTrue(
+        ex.getMessage().contains("envelope") || ex.getMessage().contains("magic byte"),
+        "short double-stripped record must be rejected; got: " + ex.getMessage()
+      );
+    }
+  }
+}
diff --git a/lib/kpipe-format-json/src/test/java/io/github/eschizoid/kpipe/format/json/JsonFormatEdgeCasesTest.java b/lib/kpipe-format-json/src/test/java/io/github/eschizoid/kpipe/format/json/JsonFormatEdgeCasesTest.java
new file mode 100644
index 00000000..acda47cb
--- /dev/null
+++ b/lib/kpipe-format-json/src/test/java/io/github/eschizoid/kpipe/format/json/JsonFormatEdgeCasesTest.java
@@ -0,0 +1,73 @@
+package io.github.eschizoid.kpipe.format.json;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+/// Payload-shape edge cases for [JsonFormat]: oversized documents, truncated UTF-8 input, and
+/// non-object top-level JSON. The format must round-trip the large case and fail
+/// loudly on the malformed cases — never quietly return a partial map.
+class JsonFormatEdgeCasesTest {
+
+  @Test
+  void oversizedDocumentRoundTrips() {
+    final var big = "y".repeat(2_000_000);
+    final var bytes = JsonFormat.INSTANCE.serialize(java.util.Map.of("blob", big));
+
+    final var decoded = JsonFormat.INSTANCE.deserialize(bytes);
+
+    assertNotNull(decoded);
+    assertEquals(big, decoded.get("blob"));
+  }
+
+  @Test
+  void truncatedMidValueThrows() {
+    // Valid prefix, cut off inside a string value — a partial parse must not be returned.
+    final var truncated = "{\"name\":\"Alic".getBytes(StandardCharsets.UTF_8);
+    assertDecodeFailure(truncated);
+  }
+
+  @Test
+  void invalidUtf8InStringValueThrows() {
+    // A lone leading byte of a 3-byte UTF-8 sequence (E2 = first byte of €) inside a string
+    // value. The byte[] (UTF-8) decode path validates the encoding and throws rather than
+    // returning a half-decoded map that downstream would treat as a processed record.
+    final var bytes = new byte[] { '{', '"', 'k', '"', ':', '"', (byte) 0xE2, '"', '}' };
+    assertDecodeFailure(bytes);
+  }
+
+  @Test
+  void topLevelArrayThrowsRatherThanReturningNull() {
+    // parseObject expects an object; a top-level array is not a Map<String, Object>. This must
+    // throw, never surface as a quietly null or empty map the pipeline would mistake for a filter.
+    final var array = "[1,2,3]".getBytes(StandardCharsets.UTF_8);
+    assertDecodeFailure(array);
+  }
+
+  @Test
+  void topLevelScalarThrowsRatherThanReturningNull() {
+    final var scalar = "42".getBytes(StandardCharsets.UTF_8);
+    assertDecodeFailure(scalar);
+  }
+
+  @Test
+  void unbalancedBracesThrow() {
+    final var unbalanced = "{\"a\":{\"b\":1}".getBytes(StandardCharsets.UTF_8);
+    assertDecodeFailure(unbalanced);
+  }
+
+  /// Asserts the input fails as the format's own decode error — not a generic exception that would
+  /// green this test on an unrelated bug. `JsonFormat.deserialize` wraps parse failures with this
+  /// prefix; anything else (NPE, coercion) propagates without it and fails the check loudly.
+  private static void assertDecodeFailure(final byte[] data) {
+    final var ex = assertThrows(RuntimeException.class, () -> JsonFormat.INSTANCE.deserialize(data));
+    assertTrue(
+      ex.getMessage() != null && ex.getMessage().contains("JsonFormat.deserialize failed"),
+      () -> "expected a JsonFormat decode failure, got: " + ex
+    );
+  }
+}
diff --git a/lib/kpipe-format-protobuf/src/test/java/io/github/eschizoid/kpipe/format/protobuf/ProtobufFormatEdgeCasesTest.java b/lib/kpipe-format-protobuf/src/test/java/io/github/eschizoid/kpipe/format/protobuf/ProtobufFormatEdgeCasesTest.java
new file mode 100644
index 00000000..315d10c7
--- /dev/null
+++ b/lib/kpipe-format-protobuf/src/test/java/io/github/eschizoid/kpipe/format/protobuf/ProtobufFormatEdgeCasesTest.java
@@ -0,0 +1,168 @@
+package io.github.eschizoid.kpipe.format.protobuf;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.protobuf.DescriptorProtos;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import io.github.eschizoid.kpipe.registry.MessagePipeline;
+import io.github.eschizoid.kpipe.registry.MessageProcessorRegistry;
+import io.github.eschizoid.kpipe.registry.Result;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/// Protobuf edge cases: oversized payloads, truncated wire bytes mid-field, and the 6-byte
+/// single-message Confluent envelope boundary driven through the real built pipeline. An
+/// off-by-one on the envelope strips the wrong bytes and corrupts every downstream field.
+class ProtobufFormatEdgeCasesTest {
+
+  private Descriptors.Descriptor descriptor;
+  private ProtobufFormat format;
+  private MessageProcessorRegistry registry;
+
+  @BeforeEach
+  void setUp() throws Exception {
+    descriptor = buildTestDescriptor();
+    format = new ProtobufFormat(descriptor);
+    registry = new MessageProcessorRegistry();
+  }
+
+  /// Drives `bytes` through deserialize -> process -> serialize, re-throwing a failure cause.
+  private static <T> byte[] roundTrip(final MessagePipeline<T> pipeline, final byte[] bytes) {
+    final var deserialized = pipeline.deserializeOrFail(bytes);
+    return switch (pipeline.process(deserialized)) {
+      case Result.Passed<T> p -> pipeline.serialize(p.value());
+      case Result.Filtered<T> _ -> null;
+      case Result.Failed<T> f -> {
+        if (f.cause() instanceof RuntimeException re) throw re;
+        if (f.cause() instanceof Error err) throw err;
+        throw new RuntimeException(f.cause());
+      }
+    };
+  }
+
+  /// Prepends the 6-byte single-message Confluent envelope: 1-byte magic 0x00 + 4-byte
+  /// big-endian schema id + 1-byte message-index header (0x00 for the first message).
+  private static byte[] withEnvelope(final int schemaId, final byte[] payload) {
+    final var framed = new byte[6 + payload.length];
+    framed[0] = 0x00;
+    framed[1] = (byte) ((schemaId >>> 24) & 0xff);
+    framed[2] = (byte) ((schemaId >>> 16) & 0xff);
+    framed[3] = (byte) ((schemaId >>> 8) & 0xff);
+    framed[4] = (byte) (schemaId & 0xff);
+    framed[5] = 0x00;
+    System.arraycopy(payload, 0, framed, 6, payload.length);
+    return framed;
+  }
+
+  private DynamicMessage message(final long id, final String name) {
+    return DynamicMessage.newBuilder(descriptor)
+      .setField(descriptor.findFieldByName("id"), id)
+      .setField(descriptor.findFieldByName("name"), name)
+      .build();
+  }
+
+  @Test
+  void oversizedPayloadRoundTrips() {
+    final var big = "z".repeat(2_000_000);
+    final var bytes = format.serialize(message(1L, big));
+
+    final var pipeline = registry.pipeline(format).build();
+    final var result = roundTrip(pipeline, bytes);
+
+    assertNotNull(result);
+    final var decoded = format.deserialize(result);
+    assertEquals(big, decoded.getField(descriptor.findFieldByName("name")));
+  }
+
+  @Test
+  void truncatedFieldBytesThrow() {
+    // Tag byte for field 2 (name, wire type 2 = length-delimited) says "10 bytes follow" but
+    // the buffer ends early. A partial decode must throw, not return a half-built message.
+    final var truncated = new byte[] { 0x12, 0x0a, 'A', 'l', 'i' };
+    assertThrows(RuntimeException.class, () -> format.deserialize(truncated));
+  }
+
+  @Test
+  void skipBytesSixStripsTheEnvelopeAndDecodesCleanly() {
+    final var payload = format.serialize(message(7L, "framed"));
+    final var framed = withEnvelope(99, payload);
+
+    final var pipeline = registry.pipeline(format).skipBytes(6).build();
+    final var decoded = pipeline.deserializeOrFail(framed);
+
+    assertNotNull(decoded);
+    assertEquals(7L, decoded.getField(descriptor.findFieldByName("id")));
+    assertEquals("framed", decoded.getField(descriptor.findFieldByName("name")));
+  }
+
+  @Test
+  void offByOneSkipFiveCorruptsOrFails() {
+    // Leaving the message-index byte in front shifts the protobuf tag parse — corruption.
+    final var payload = format.serialize(message(7L, "framed"));
+    final var framed = withEnvelope(99, payload);
+
+    final var pipeline = registry.pipeline(format).skipBytes(5).build();
+
+    var corrupted = false;
+    try {
+      final var decoded = pipeline.deserializeOrFail(framed);
+      corrupted =
+        decoded == null ||
+        !"framed".equals(decoded.getField(descriptor.findFieldByName("name"))) ||
+        !Long.valueOf(7L).equals(decoded.getField(descriptor.findFieldByName("id")));
+    } catch (final RuntimeException e) {
+      // A throw is an acceptable corruption signal, but require the protobuf decoder's own
+      // failure — not an unrelated error that would green this test vacuously.
+      assertTrue(
+        e.getMessage() != null && e.getMessage().contains("ProtobufFormat.deserialize failed"),
+        () -> "expected a protobuf decode failure, got: " + e
+      );
+      corrupted = true;
+    }
+    assertTrue(corrupted, "skip(5) must not decode to the clean message");
+  }
+
+  @Test
+  void emptyMessageSerializesToZeroLengthAndDeserializeTreatsItAsNull() {
+    // A proto3 message with all-default fields encodes to zero bytes. Serialize yields an
+    // empty array; the empty array is the null/empty deserialize case, not a parse error.
+    final var empty = DynamicMessage.newBuilder(descriptor).build();
+    final var bytes = format.serialize(empty);
+
+    assertArrayEquals(new byte[0], bytes, "an all-default proto3 message encodes to zero bytes");
+    assertNull(format.deserialize(bytes));
+  }
+
+  private static Descriptors.Descriptor buildTestDescriptor() throws Descriptors.DescriptorValidationException {
+    final var msg = DescriptorProtos.DescriptorProto.newBuilder()
+      .setName("TestMessage")
+      .addField(field("id", 1, DescriptorProtos.FieldDescriptorProto.Type.TYPE_INT64))
+      .addField(field("name", 2, DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING))
+      .build();
+
+    final var fileProto = DescriptorProtos.FileDescriptorProto.newBuilder()
+      .setName("test.proto")
+      .setPackage("test")
+      .setSyntax("proto3")
+      .addMessageType(msg)
+      .build();
+
+    return Descriptors.FileDescriptor.buildFrom(fileProto, new Descriptors.FileDescriptor[0]).findMessageTypeByName(
+      "TestMessage"
+    );
+  }
+
+  private static DescriptorProtos.FieldDescriptorProto field(
+    final String name,
+    final int number,
+    final DescriptorProtos.FieldDescriptorProto.Type type
+  ) {
+    return DescriptorProtos.FieldDescriptorProto.newBuilder().setName(name).setNumber(number).setType(type).build();
+  }
+}
diff --git a/lib/kpipe-producer/src/test/java/io/github/eschizoid/kpipe/producer/KPipeProducerDurabilityTest.java b/lib/kpipe-producer/src/test/java/io/github/eschizoid/kpipe/producer/KPipeProducerDurabilityTest.java
new file mode 100644
index 00000000..d0153a34
--- /dev/null
+++ b/lib/kpipe-producer/src/test/java/io/github/eschizoid/kpipe/producer/KPipeProducerDurabilityTest.java
@@ -0,0 +1,195 @@
+package io.github.eschizoid.kpipe.producer;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.github.eschizoid.kpipe.producer.tracing.Tracer;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.errors.NetworkException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.jupiter.api.Test;
+
+/// Durability and configuration-passthrough coverage for [KPipeProducer], driven by a real
+/// [MockProducer] rather than a Mockito stub.
+///
+/// The Mockito-based tests in [KPipeProducerTest] verify the return value against pre-resolved
+/// `CompletableFuture` stubs. These tests use [MockProducer] in manual-completion mode so the
+/// broker ack future is resolved (or errored) explicitly, exercising the same code path a live
+/// broker drives: `producer.send(record).get()` blocks until the ack lands.
+///
+/// The load-bearing property: `sendToDlq` returns `true` only when the record's ack future
+/// completes without error. The consumer commits the source offset solely on a `true` return;
+/// a `true` on a non-durable send would silently lose the record on the source topic. A `false`
+/// leaves the offset pending so the record is reprocessed.
+class KPipeProducerDurabilityTest {
+
+  private static final String TOPIC = "source-topic";
+  private static final String DLQ_TOPIC = "source-topic-dlq";
+
+  private static MockProducer<byte[], byte[]> manualMockProducer() {
+    // autoComplete=false → send() returns an incomplete future; the test drives completeNext /
+    // errorNext to resolve it, mirroring a broker that acks (or fails) asynchronously.
+    return new MockProducer<>(false, (Partitioner) null, new ByteArraySerializer(), new ByteArraySerializer());
+  }
+
+  private static ConsumerRecord<byte[], byte[]> failedRecord() {
+    return new ConsumerRecord<>(TOPIC, 0, 7L, "k".getBytes(), "v".getBytes());
+  }
+
+  /// A completed ack future must surface as `true`, and the record must reach the underlying
+  /// producer. The result is undecided until the ack lands.
+  @Test
+  void sendToDlqReturnsTrueOnlyAfterDurableAck() throws Exception {
+    final var mock = manualMockProducer();
+    final var producer = new KPipeProducer<>(mock, false, null, Tracer.noop());
+
+    final var done = new CountDownLatch(1);
+    final var result = new boolean[1];
+    final var thread = Thread.ofVirtual().start(() -> {
+      result[0] = producer.sendToDlq(DLQ_TOPIC, failedRecord(), TOPIC, new RuntimeException("boom"));
+      done.countDown();
+    });
+
+    // The send is parked on get(); the ack future is not yet complete, so the result is undecided.
+    awaitBufferedSend(mock);
+    assertEquals(1, done.getCount(), "sendToDlq must not return before the broker ack lands");
+
+    assertTrue(mock.completeNext(), "MockProducer should have one buffered send to complete");
+    assertTrue(done.await(5, TimeUnit.SECONDS), "sendToDlq must return once the ack completes");
+    thread.join();
+
+    assertTrue(result[0], "a completed ack must surface as true");
+    assertEquals(1, mock.history().size(), "the record must have reached the underlying producer");
+    assertEquals(DLQ_TOPIC, mock.history().getFirst().topic());
+  }
+
+  /// A broker rejection must surface as `false` so the caller leaves the source offset pending.
+  @Test
+  void sendToDlqReturnsFalseOnSendError() throws Exception {
+    final var mock = manualMockProducer();
+    final var producer = new KPipeProducer<>(mock, false, null, Tracer.noop());
+
+    final var done = new CountDownLatch(1);
+    final var result = new boolean[] { true };
+    Thread.ofVirtual().start(() -> {
+      result[0] = producer.sendToDlq(DLQ_TOPIC, failedRecord(), TOPIC, new RuntimeException("boom"));
+      done.countDown();
+    });
+
+    awaitBufferedSend(mock);
+    assertTrue(mock.errorNext(new TimeoutException("broker unreachable")), "buffered send should error");
+
+    assertTrue(done.await(5, TimeUnit.SECONDS), "sendToDlq must return once the ack future errors");
+    assertFalse(result[0], "a failed ack must surface as false so the offset is held");
+  }
+
+  /// A transient failure that the application retries by calling `sendToDlq` again must end in a
+  /// `true` once the retry's ack lands. The first attempt errors; the second succeeds. This mirrors
+  /// the consumer reprocessing a held offset after a transient DLQ outage.
+  @Test
+  void sendToDlqRetryAfterTransientFailureEventuallySucceeds() throws Exception {
+    final var mock = manualMockProducer();
+    final var producer = new KPipeProducer<>(mock, false, null, Tracer.noop());
+
+    // Attempt 1: transient failure → false.
+    final var firstDone = new CountDownLatch(1);
+    final var firstResult = new boolean[] { true };
+    Thread.ofVirtual().start(() -> {
+      firstResult[0] = producer.sendToDlq(DLQ_TOPIC, failedRecord(), TOPIC, new RuntimeException("boom"));
+      firstDone.countDown();
+    });
+    awaitBufferedSend(mock);
+    assertTrue(mock.errorNext(new NetworkException("transient blip")), "first send should error");
+    assertTrue(firstDone.await(5, TimeUnit.SECONDS));
+    assertFalse(firstResult[0], "transient failure must surface as false");
+
+    // Attempt 2: retry of the same record completes durably → true.
+    final var secondDone = new CountDownLatch(1);
+    final var secondResult = new boolean[1];
+    Thread.ofVirtual().start(() -> {
+      secondResult[0] = producer.sendToDlq(DLQ_TOPIC, failedRecord(), TOPIC, new RuntimeException("boom"));
+      secondDone.countDown();
+    });
+    awaitBufferedSend(mock);
+    assertTrue(mock.completeNext(), "retry send should complete");
+    assertTrue(secondDone.await(5, TimeUnit.SECONDS));
+
+    assertTrue(secondResult[0], "the durable retry must surface as true");
+  }
+
+  /// `send` (the synchronous path the DLQ uses) returns the broker-assigned metadata once the ack
+  /// lands and rethrows as a wrapped failure when the ack errors — the two outcomes `sendToDlq`
+  /// keys its boolean off of.
+  @Test
+  void synchronousSendSurfacesAckOutcome() throws Exception {
+    final var mock = manualMockProducer();
+    final var producer = new KPipeProducer<>(mock, false, null, Tracer.noop());
+
+    final var done = new CountDownLatch(1);
+    final var thrown = new Throwable[1];
+    Thread.ofVirtual().start(() -> {
+      try {
+        producer.send(new ProducerRecord<>(TOPIC, "k".getBytes(), "v".getBytes()));
+      } catch (final Throwable t) {
+        thrown[0] = t;
+      }
+      done.countDown();
+    });
+
+    awaitBufferedSend(mock);
+    assertTrue(mock.errorNext(new TimeoutException("no ack")), "buffered send should error");
+    assertTrue(done.await(5, TimeUnit.SECONDS));
+
+    assertNotNull(thrown[0], "an errored ack must propagate as a thrown failure");
+    assertEquals("Send failed", thrown[0].getMessage());
+  }
+
+  /// The builder forwards reliability config (`enable.idempotence`, `acks`, in-flight cap, retries)
+  /// untouched to the underlying Kafka producer properties. The wrapper only fills serializer and
+  /// `client.id` defaults; it must never silently downgrade the durability guarantees a caller set.
+  @Test
+  void builderForwardsIdempotenceAndAcksConfigUntouched() {
+    final var userProps = new Properties();
+    userProps.setProperty("bootstrap.servers", "broker:9092");
+    userProps.setProperty("enable.idempotence", "true");
+    userProps.setProperty("acks", "all");
+    userProps.setProperty("max.in.flight.requests.per.connection", "5");
+    userProps.setProperty("retries", "2147483647");
+
+    final var built = KPipeProducer.Builder.buildProducerProperties(userProps);
+
+    assertEquals("true", built.getProperty("enable.idempotence"), "idempotence flag must pass through");
+    assertEquals("all", built.getProperty("acks"), "acks must not be downgraded");
+    assertEquals("5", built.getProperty("max.in.flight.requests.per.connection"));
+    assertEquals("2147483647", built.getProperty("retries"));
+  }
+
+  /// The wrapper does not inject an `acks` or `enable.idempotence` default of its own — absence in
+  /// means absence out, so the Kafka client's own defaults apply and the wrapper never masks them.
+  @Test
+  void builderDoesNotInjectReliabilityDefaults() {
+    final var userProps = new Properties();
+    userProps.setProperty("bootstrap.servers", "broker:9092");
+
+    final var built = KPipeProducer.Builder.buildProducerProperties(userProps);
+
+    assertNull(built.getProperty("acks"), "wrapper must not invent an acks default");
+    assertNull(built.getProperty("enable.idempotence"), "wrapper must not invent an idempotence default");
+  }
+
+  /// Spin until the send is buffered in the MockProducer so the test can deterministically drive
+  /// completeNext / errorNext. The sending virtual thread is parked on get() at that point.
+  private static void awaitBufferedSend(final MockProducer<byte[], byte[]> mock) throws InterruptedException {
+    final var deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+    while (mock.flushed() && System.nanoTime() < deadline) {
+      Thread.sleep(2);
+    }
+    assertFalse(mock.flushed(), "the send should have been buffered for completion");
+  }
+}