diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/BackpressureHysteresisPropertyTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/BackpressureHysteresisPropertyTest.java
new file mode 100644
index 00000000..eae6d854
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/BackpressureHysteresisPropertyTest.java
@@ -0,0 +1,98 @@
+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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.github.eschizoid.kpipe.consumer.BackpressureController.Action;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import net.jqwik.api.Arbitraries;
+import net.jqwik.api.Arbitrary;
+import net.jqwik.api.ForAll;
+import net.jqwik.api.Property;
+import net.jqwik.api.Provide;
+import net.jqwik.api.constraints.LongRange;
+
+/// Property-based coverage for the backpressure watermark hysteresis. Where
+/// `WatermarkHysteresisTest` pins the exact watermark edges with hand-picked values, this suite
+/// generates randomized load walks — arbitrary in-flight values over arbitrary watermark pairs
+/// (low fixed at 70% of high, the shape of the production defaults) — and asserts the decision
+/// invariants hold for EVERY generated walk:
+///
+///   - PAUSE never fires while the metric is below the high watermark;
+///   - RESUME never fires while the metric is above the low watermark;
+///   - PAUSE and RESUME strictly alternate — no pause-pause or resume-resume without the
+///     opposite edge between them, so the band between the watermarks is a true dead zone.
+///
+/// Each walk drives [BackpressureController#check] through the same paused-flag feedback loop
+/// the consumer runs: every decision is applied to the flag that feeds the next check, exactly
+/// like `tickBackpressure`.
+class BackpressureHysteresisPropertyTest {
+
+  @Provide
+  Arbitrary<List<Long>> loadWalks() {
+    return Arbitraries.longs().between(0L, 20_000L).list().ofMinSize(1).ofMaxSize(200);
+  }
+
+  @Property(tries = 200)
+  void pauseOnlyAtOrAboveHighAndResumeOnlyAtOrBelowLow(
+    @ForAll("loadWalks") final List<Long> loads,
+    @ForAll @LongRange(min = 100, max = 10_000) final long high
+  ) {
+    final var low = Math.round(high * 0.7);
+    final var metric = new AtomicLong();
+    final var controller = new BackpressureController(high, low, BackpressureController.inFlightStrategy(metric::get));
+
+    var paused = false;
+    for (final var load : loads) {
+      metric.set(load);
+      switch (controller.check(null, paused)) {
+        case PAUSE -> {
+          assertTrue(load >= high, "PAUSE at %d below high watermark %d".formatted(load, high));
+          assertFalse(paused, "PAUSE must never fire while already paused");
+          paused = true;
+        }
+        case RESUME -> {
+          assertTrue(load <= low, "RESUME at %d above low watermark %d".formatted(load, low));
+          assertTrue(paused, "RESUME must never fire while running");
+          paused = false;
+        }
+        case NONE -> {
+        }
+      }
+    }
+  }
+
+  @Property(tries = 200)
+  void pauseAndResumeStrictlyAlternate(
+    @ForAll("loadWalks") final List<Long> loads,
+    @ForAll @LongRange(min = 100, max = 10_000) final long high
+  ) {
+    final var low = Math.round(high * 0.7);
+    final var metric = new AtomicLong();
+    final var controller = new BackpressureController(high, low, BackpressureController.inFlightStrategy(metric::get));
+
+    var paused = false;
+    Action lastEdge = null;
+    for (final var load : loads) {
+      metric.set(load);
+      final var action = controller.check(null, paused);
+      switch (action) {
+        case PAUSE -> {
+          assertNotEquals(Action.PAUSE, lastEdge, "pause-pause without a resume between is flapping");
+          paused = true;
+          lastEdge = action;
+        }
+        case RESUME -> {
+          assertEquals(Action.PAUSE, lastEdge, "the first RESUME edge must follow a PAUSE edge");
+          paused = false;
+          lastEdge = action;
+        }
+        case NONE -> {
+        }
+      }
+    }
+  }
+}
diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CircuitBreakerPropertyTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CircuitBreakerPropertyTest.java
new file mode 100644
index 00000000..1abae873
--- /dev/null
+++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CircuitBreakerPropertyTest.java
@@ -0,0 +1,330 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import net.jqwik.api.Arbitraries;
+import net.jqwik.api.Arbitrary;
+import net.jqwik.api.ForAll;
+import net.jqwik.api.Property;
+import net.jqwik.api.Provide;
+import net.jqwik.api.constraints.DoubleRange;
+import net.jqwik.api.constraints.IntRange;
+
+/// Property-based coverage for the circuit-breaker state machine hosted by
+/// [ConsumerHealthController]. Where `CircuitBreakerTransitionTest` pins specific transition
+/// scenarios example-by-example, this suite generates randomized event sequences — arbitrary
+/// success/failure outcomes interleaved with probe-timer firings, over arbitrary window sizes
+/// and failure thresholds — and asserts the machine invariants hold for EVERY generated stream:
+///
+///   - only legal edges are ever taken: CLOSED → OPEN, OPEN → HALF_OPEN,
+///     HALF_OPEN → CLOSED, HALF_OPEN → OPEN — never CLOSED → HALF_OPEN or OPEN → CLOSED;
+///   - the breaker never trips from CLOSED before `windowSize` outcomes fill a fresh window;
+///   - outcomes recorded while OPEN are ignored — state, transition log, and probe count all
+///     stay untouched until the probe timer fires.
+///
+/// No wall clock is involved: the OPEN → HALF_OPEN probe is captured by a hand-fired scheduler
+/// (the same seam `CircuitBreakerTransitionTest` uses), so "open-duration elapses" is an
+/// explicit generated event, and the ignored-while-OPEN invariant is checked across
+/// arbitrarily long outcome runs with the breaker deterministically held OPEN.
+class CircuitBreakerPropertyTest {
+
+  /// One generated step: a per-record outcome, or the probe timer elapsing.
+  private enum Event {
+    SUCCESS,
+    FAILURE,
+    FIRE_PROBE,
+  }
+
+  private static final Set<List<CircuitBreakerState>> LEGAL_EDGES = Set.of(
+    List.of(CircuitBreakerState.CLOSED, CircuitBreakerState.OPEN),
+    List.of(CircuitBreakerState.OPEN, CircuitBreakerState.HALF_OPEN),
+    List.of(CircuitBreakerState.HALF_OPEN, CircuitBreakerState.CLOSED),
+    List.of(CircuitBreakerState.HALF_OPEN, CircuitBreakerState.OPEN)
+  );
+
+  @Provide
+  Arbitrary<List<Event>> eventSequences() {
+    return Arbitraries.of(Event.SUCCESS, Event.FAILURE, Event.FIRE_PROBE).list().ofMinSize(1).ofMaxSize(80);
+  }
+
+  @Property(tries = 200)
+  void onlyLegalEdgesEverObserved(
+    @ForAll("eventSequences") final List<Event> events,
+    @ForAll @IntRange(min = 2, max = 10) final int windowSize,
+    @ForAll @DoubleRange(min = 0.3, max = 0.9) final double threshold
+  ) {
+    final var observer = new RecordingObserver();
+    final var scheduler = new CapturingScheduler();
+    final var hc = newController(threshold, windowSize, observer, scheduler);
+
+    for (final var event : events) apply(hc, scheduler, event);
+
+    // Prepend the initial state; the observer log then yields every consecutive edge taken.
+    final var states = new ArrayList<CircuitBreakerState>();
+    states.add(CircuitBreakerState.CLOSED);
+    states.addAll(observer.stateChanges);
+    for (var i = 1; i < states.size(); i++) {
+      final var edge = List.of(states.get(i - 1), states.get(i));
+      assertTrue(LEGAL_EDGES.contains(edge), "illegal transition %s in %s".formatted(edge, states));
+    }
+  }
+
+  @Property(tries = 200)
+  void neverTripsBeforeWindowFillsFromAFreshWindow(
+    @ForAll("eventSequences") final List<Event> events,
+    @ForAll @IntRange(min = 2, max = 10) final int windowSize,
+    @ForAll @DoubleRange(min = 0.3, max = 0.9) final double threshold
+  ) {
+    final var observer = new RecordingObserver();
+    final var scheduler = new CapturingScheduler();
+    final var hc = newController(threshold, windowSize, observer, scheduler);
+
+    // Model of the rolling window's fill level: outcomes accepted (state != OPEN) since the
+    // last reset. HALF_OPEN → CLOSED resets the window; a trip does not.
+    var samplesSinceReset = 0;
+    for (final var event : events) {
+      final var before = hc.circuitBreakerState();
+      if (event != Event.FIRE_PROBE && before != CircuitBreakerState.OPEN) samplesSinceReset++;
+      apply(hc, scheduler, event);
+      final var after = hc.circuitBreakerState();
+      if (before == CircuitBreakerState.CLOSED && after == CircuitBreakerState.OPEN) {
+        assertTrue(
+          samplesSinceReset >= windowSize,
+          "tripped after only %d of %d window samples".formatted(samplesSinceReset, windowSize)
+        );
+      }
+      if (before == CircuitBreakerState.HALF_OPEN && after == CircuitBreakerState.CLOSED) samplesSinceReset = 0;
+    }
+  }
+
+  @Property(tries = 200)
+  void outcomesWhileOpenAreIgnored(
+    @ForAll("eventSequences") final List<Event> events,
+    @ForAll @IntRange(min = 2, max = 10) final int windowSize,
+    @ForAll @DoubleRange(min = 0.3, max = 0.9) final double threshold
+  ) {
+    final var observer = new RecordingObserver();
+    final var scheduler = new CapturingScheduler();
+    final var hc = newController(threshold, windowSize, observer, scheduler);
+
+    for (final var event : events) {
+      final var before = hc.circuitBreakerState();
+      final var transitionsBefore = observer.stateChanges.size();
+      final var probesBefore = scheduler.scheduled.size();
+      apply(hc, scheduler, event);
+      if (event != Event.FIRE_PROBE && before == CircuitBreakerState.OPEN) {
+        assertSame(CircuitBreakerState.OPEN, hc.circuitBreakerState(), "an outcome while OPEN must not move the state");
+        assertEquals(transitionsBefore, observer.stateChanges.size(), "no transition may fire from an OPEN outcome");
+        assertEquals(probesBefore, scheduler.scheduled.size(), "no extra probe may be armed by an OPEN outcome");
+      }
+    }
+  }
+
+  // ─────────────────────────── Drive helpers ────────────────────────────────
+
+  private static ConsumerHealthController newController(
+    final double threshold,
+    final int windowSize,
+    final RecordingObserver observer,
+    final CapturingScheduler scheduler
+  ) {
+    final var cb = new CircuitBreakerController(threshold, windowSize, Duration.ofMillis(300));
+    return new ConsumerHealthController(null, cb, scheduler, observer, observer);
+  }
+
+  private static void apply(final ConsumerHealthController hc, final CapturingScheduler scheduler, final Event event) {
+    switch (event) {
+      case SUCCESS -> hc.recordOutcome(true);
+      case FAILURE -> hc.recordOutcome(false);
+      // Firing an already-fired one-shot is a clean lost-CAS no-op, so no state guard is needed
+      // beyond "a probe was armed at some point."
+      case FIRE_PROBE -> {
+        if (!scheduler.scheduled.isEmpty()) scheduler.fireLatest();
+      }
+    }
+  }
+
+  /// Records pause/resume and every circuit-breaker state change so properties can assert on the
+  /// exact transition sequence. Same recording pattern as `CircuitBreakerTransitionTest`.
+  private static final class RecordingObserver
+    implements ConsumerHealthController.PauseLifecycleHook, ConsumerHealthController.HealthMetricsObserver
+  {
+
+    final List<CircuitBreakerState> stateChanges = new CopyOnWriteArrayList<>();
+
+    @Override
+    public void onPause() {}
+
+    @Override
+    public void onResume() {}
+
+    @Override
+    public void onBackpressurePause() {}
+
+    @Override
+    public void onBackpressureTimeMs(final long ms) {}
+
+    @Override
+    public void onCircuitBreakerTrip() {}
+
+    @Override
+    public void onCircuitBreakerStateChange(final CircuitBreakerState state) {
+      stateChanges.add(state);
+    }
+
+    @Override
+    public void onCircuitBreakerTimeOpenMs(final long ms) {}
+  }
+
+  /// A scheduler that never runs anything on its own: each submitted probe task is captured so
+  /// the property fires it by hand, making "open-duration elapsed" a deterministic event.
+  private static final class CapturingScheduler implements ScheduledExecutorService {
+
+    final List<Runnable> scheduled = new ArrayList<>();
+
+    void fireLatest() {
+      scheduled.getLast().run();
+    }
+
+    @Override
+    public ScheduledFuture<?> schedule(final Runnable command, final long delay, final TimeUnit unit) {
+      scheduled.add(command);
+      return new NoopFuture();
+    }
+
+    private static final class NoopFuture implements ScheduledFuture<Object> {
+
+      @Override
+      public long getDelay(final TimeUnit unit) {
+        return 0;
+      }
+
+      @Override
+      public int compareTo(final Delayed o) {
+        return 0;
+      }
+
+      @Override
+      public boolean cancel(final boolean mayInterruptIfRunning) {
+        return true;
+      }
+
+      @Override
+      public boolean isCancelled() {
+        return false;
+      }
+
+      @Override
+      public boolean isDone() {
+        return false;
+      }
+
+      @Override
+      public Object get() {
+        return null;
+      }
+
+      @Override
+      public Object get(final long timeout, final TimeUnit unit) {
+        return null;
+      }
+    }
+
+    // ── Unused ScheduledExecutorService surface ─────────────────────────────
+    @Override
+    public <V> ScheduledFuture<V> schedule(final Callable<V> c, final long d, final TimeUnit u) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(final Runnable c, final long i, final long p, final TimeUnit u) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(final Runnable c, final long i, final long d, final TimeUnit u) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void shutdown() {}
+
+    @Override
+    public List<Runnable> shutdownNow() {
+      return List.of();
+    }
+
+    @Override
+    public boolean isShutdown() {
+      return false;
+    }
+
+    @Override
+    public boolean isTerminated() {
+      return false;
+    }
+
+    @Override
+    public boolean awaitTermination(final long timeout, final TimeUnit unit) {
+      return true;
+    }
+
+    @Override
+    public <T> Future<T> submit(final Callable<T> task) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> Future<T> submit(final Runnable task, final T result) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Future<?> submit(final Runnable task) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(
+      final Collection<? extends Callable<T>> tasks,
+      final long timeout,
+      final TimeUnit unit
+    ) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> T invokeAny(final Collection<? extends Callable<T>> tasks) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> T invokeAny(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void execute(final Runnable command) {
+      throw new UnsupportedOperationException();
+    }
+  }
+}
diff --git a/lib/kpipe-test/src/main/java/io/github/eschizoid/kpipe/test/TestStream.java b/lib/kpipe-test/src/main/java/io/github/eschizoid/kpipe/test/TestStream.java
index 9cfb7cd0..e2791f0a 100644
--- a/lib/kpipe-test/src/main/java/io/github/eschizoid/kpipe/test/TestStream.java
+++ b/lib/kpipe-test/src/main/java/io/github/eschizoid/kpipe/test/TestStream.java
@@ -206,9 +206,20 @@ private void ensureOpen() {
     if (closed.get()) throw new IllegalStateException("TestStream is closed");
   }
 
-  /// Builder for [TestStream]. Mirrors the production facade's `pipe` / `filter` / `peek` /
-  /// `toCustom` vocabulary so a passing TestStream chain translates 1:1 to a `KPipe.X(...)`
-  /// chain.
+  /// Builder for [TestStream]. The operator vocabulary — `pipe` / `filter` / `peek` / `when` —
+  /// and the sink terminals (`toCustom`, `toBatch`) map 1:1 onto the production facade, so a
+  /// transform chain proven here carries over to a `KPipe.X(...)` chain unchanged.
+  ///
+  /// Deliberate differences from the facade's `Stream`:
+  ///
+  ///   * **Mutable builder.** Methods return `this`, not a new immutable instance — branching
+  ///     one builder into two streams is not supported; build one `TestStream` per chain.
+  ///   * **`toBatch(sink, maxSize)` is size-only.** The facade takes a `BatchPolicy`
+  ///     (size + age); the test kit's v1 batch semantics flush on size and shutdown only, so
+  ///     the age trigger is not configurable here.
+  ///   * **No observers.** `onFiltered` / `onFailed` / `peekResult` do not exist — assert
+  ///     filter and failure outcomes through the captured sink output,
+  ///     [TestStream#metrics], and [TestStream#errors] instead.
   ///
   /// @param <T> the pipeline value type
   public static final class Builder<T> {
@@ -268,6 +279,22 @@ public Builder<T> peek(final Consumer<T> sideEffect) {
       return this;
     }
 
+    /// Appends a conditional transform: records matching the predicate go through `ifTrue`,
+    /// the rest through `ifFalse`. Either branch may return `null` to filter the record. Same
+    /// composition as the production facade's `when`.
+    ///
+    /// @param cond the branching predicate
+    /// @param ifTrue applied when the predicate matches
+    /// @param ifFalse applied when the predicate does not match
+    /// @return this builder
+    public Builder<T> when(final Predicate<T> cond, final UnaryOperator<T> ifTrue, final UnaryOperator<T> ifFalse) {
+      Objects.requireNonNull(cond, "condition cannot be null");
+      Objects.requireNonNull(ifTrue, "ifTrue cannot be null");
+      Objects.requireNonNull(ifFalse, "ifFalse cannot be null");
+      operators.add(value -> cond.test(value) ? ifTrue.apply(value) : ifFalse.apply(value));
+      return this;
+    }
+
     /// Sets the terminal sink — typically a [CapturingSink]. Mutually exclusive with [#toBatch].
     ///
     /// @param sink the terminal sink
diff --git a/lib/kpipe-test/src/test/java/io/github/eschizoid/kpipe/test/TestStreamTest.java b/lib/kpipe-test/src/test/java/io/github/eschizoid/kpipe/test/TestStreamTest.java
index 1ac938fd..ee155700 100644
--- a/lib/kpipe-test/src/test/java/io/github/eschizoid/kpipe/test/TestStreamTest.java
+++ b/lib/kpipe-test/src/test/java/io/github/eschizoid/kpipe/test/TestStreamTest.java
@@ -149,6 +149,61 @@ void peekObservesWithoutMutating() {
     }
   }
 
+  @Test
+  void whenRoutesThroughMatchingBranch() {
+    final UnaryOperator<Map<String, Object>> tagActive = m -> {
+      final var copy = new HashMap<>(m);
+      copy.put("branch", "active");
+      return copy;
+    };
+    final UnaryOperator<Map<String, Object>> tagInactive = m -> {
+      final var copy = new HashMap<>(m);
+      copy.put("branch", "inactive");
+      return copy;
+    };
+
+    final var captured = new CapturingSink<Map<String, Object>>();
+    try (
+      final var driver = TestStream.<Map<String, Object>>builder(JsonFormat.INSTANCE)
+        .when(ACTIVE, tagActive, tagInactive)
+        .toCustom(captured)
+        .build()
+    ) {
+      driver.send(record("a", true));
+      driver.send(record("b", false));
+      driver.flush();
+
+      assertEquals(
+        List.of("active", "inactive"),
+        captured
+          .captured()
+          .stream()
+          .map(m -> m.get("branch"))
+          .toList(),
+        "when must route each record through exactly the branch its predicate selects"
+      );
+    }
+  }
+
+  @Test
+  void whenBranchReturningNullFiltersTheRecord() {
+    final var captured = new CapturingSink<Map<String, Object>>();
+    try (
+      final var driver = TestStream.<Map<String, Object>>builder(JsonFormat.INSTANCE)
+        .when(ACTIVE, UnaryOperator.identity(), m -> null)
+        .toCustom(captured)
+        .build()
+    ) {
+      driver.send(record("keep", true));
+      driver.send(record("drop", false));
+      driver.flush();
+
+      assertEquals(List.of("keep"), ids(captured), "a null-returning branch filters the record");
+      assertEquals(2L, driver.metrics().get("messagesProcessed"), "branch-filtered records still count processed");
+      assertEquals(List.of(), driver.errors(), "branch filtering is not an error");
+    }
+  }
+
   // ────────────────────────────────────────────────────────────────────────────────
   // Filter path — dropped records are processed, not errors, never reach the sink
   // ────────────────────────────────────────────────────────────────────────────────