fix: truthful health signal — isHealthy no longer lies during a breaker trip

examples/demo/src/main/java/io/github/eschizoid/kpipe/demo/DemoApp.java

@@ -8,6 +8,7 @@
 import io.github.eschizoid.kpipe.format.avro.AvroFormat;
 import io.github.eschizoid.kpipe.format.json.JsonConsoleSink;
 import io.github.eschizoid.kpipe.format.protobuf.ProtobufFormat;
+import io.github.eschizoid.kpipe.health.HttpHealthServer;
 import io.github.eschizoid.kpipe.metrics.otel.OtelConsumerMetrics;
 import io.github.eschizoid.kpipe.registry.Operators;
 import io.github.eschizoid.kpipe.schemaregistry.confluent.ConfluentSchemaResolver;
@@ -50,7 +51,20 @@ static void main() {
 
     try (final var app = new DemoApp(config, avroFormat, protoFormat)) {
       LOGGER.log(Level.INFO, "Demo application started — JSON/Avro/Protobuf routes via KPipe.multi");
-      app.handle.awaitShutdown();
+      // Liveness probe wired to the handle's health snapshot: reports unhealthy when the
+      // circuit breaker is OPEN, not merely when the process has exited. Enabled/configured via
+      // HEALTH_* environment variables (see HealthConfig); disabled = empty Optional, no server.
+      final var healthServer = HttpHealthServer.fromEnv(
+        () -> app.handle.health().healthy(),
+        () -> app.handle.health().inFlight(),
+        () -> app.handle.health().paused(),
+        "demo-app"
+      );
+      try {
+        app.handle.awaitShutdown();
+      } finally {
+        healthServer.ifPresent(HttpHealthServer::close);
+      }
     } catch (final Exception e) {
       LOGGER.log(Level.ERROR, "Fatal error in demo application", e);
       System.exit(1);

lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultHandle.java

@@ -1,5 +1,6 @@
 package io.github.eschizoid.kpipe;
 
+import io.github.eschizoid.kpipe.consumer.HealthSnapshot;
 import io.github.eschizoid.kpipe.consumer.KPipeConsumer;
 import java.time.Duration;
 import java.util.List;
@@ -16,7 +17,15 @@ record DefaultHandle(KPipeConsumer consumer) implements Handle {
 
   @Override
   public boolean isHealthy() {
-    return consumer.isRunning();
+    // Delegates to the snapshot's definition: running AND breaker not OPEN. `isRunning()` alone
+    // reads true while a tripped circuit breaker has the consumer PAUSED — the one incident a
+    // liveness probe must see.
+    return consumer.health().healthy();
+  }
+
+  @Override
+  public HealthSnapshot health() {
+    return consumer.health();
   }
 
   @Override

lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/Handle.java

@@ -1,5 +1,6 @@
 package io.github.eschizoid.kpipe;
 
+import io.github.eschizoid.kpipe.consumer.HealthSnapshot;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -15,6 +16,13 @@ public interface Handle extends AutoCloseable {
   /// @return true when the consumer is running and the configured health check passes
   boolean isHealthy();
 
+  /// Full health snapshot: running/paused state, active pause sources, circuit-breaker state,
+  /// and in-flight count. Wire liveness probes (e.g. `HttpHealthServer`) to this rather than
+  /// polling `metrics()`.
+  ///
+  /// @return an immutable point-in-time health view
+  HealthSnapshot health();
+
   /// Returns an unmodifiable snapshot of the consumer's metrics. All values are counters or
   /// gauges represented as `Long`. Returns an empty map when metrics are disabled on the
   /// underlying consumer.

lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/HandleDefaultsTest.java

@@ -3,9 +3,12 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.*;
 
+import io.github.eschizoid.kpipe.consumer.CircuitBreakerState;
+import io.github.eschizoid.kpipe.consumer.HealthSnapshot;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.Test;
 
@@ -26,6 +29,11 @@ public boolean isHealthy() {
       return true;
     }
 
+    @Override
+    public HealthSnapshot health() {
+      return new HealthSnapshot(true, false, Set.of(), CircuitBreakerState.CLOSED, 0L);
+    }
+
     @Override
     public Map<String, Long> metrics() {
       return Map.of();

lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/HealthSnapshot.java

@@ -0,0 +1,35 @@
+package io.github.eschizoid.kpipe.consumer;
+
+import java.util.Set;
+
+/// Point-in-time health view of a running consumer, taken via [KPipeConsumer#health()].
+///
+/// This is the one place the "is this consumer healthy" question is answered. A consumer that
+/// paused itself because the circuit breaker tripped is alive but NOT healthy — the breaker
+/// pauses consumption precisely because downstream is failing — so [#healthy()] is the value a
+/// liveness/readiness probe should serve, not `running` alone.
+///
+/// @param running        the consumer thread is alive (RUNNING or PAUSED state)
+/// @param paused         consumption is currently paused (any source)
+/// @param pauseSources   names of the active pause sources (`MANUAL`, `BACKPRESSURE`,
+///                       `CIRCUIT_BREAKER`); empty when not paused
+/// @param circuitBreaker current circuit-breaker state; `CLOSED` when no breaker is configured
+/// @param inFlight       records currently in flight (dispatched + buffered batch records)
+public record HealthSnapshot(
+  boolean running,
+  boolean paused,
+  Set<String> pauseSources,
+  CircuitBreakerState circuitBreaker,
+  long inFlight
+) {
+  public HealthSnapshot {
+    pauseSources = Set.copyOf(pauseSources);
+  }
+
+  /// `true` when the consumer is running and the circuit breaker is not OPEN. Backpressure or
+  /// manual pauses do not make a consumer unhealthy — they are normal flow control — but an
+  /// OPEN breaker means downstream is failing and consumption is deliberately halted.
+  public boolean healthy() {
+    return running && circuitBreaker != CircuitBreakerState.OPEN;
+  }
+}

lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java

@@ -1,15 +1,12 @@
 package io.github.eschizoid.kpipe.consumer;
 
-import io.github.eschizoid.kpipe.consumer.config.AppConfig;
 import io.github.eschizoid.kpipe.metrics.ConsumerMetricKeys;
 import io.github.eschizoid.kpipe.metrics.ConsumerMetrics;
 import io.github.eschizoid.kpipe.metrics.KPipeMetricsReporter;
 import io.github.eschizoid.kpipe.producer.KPipeProducer;
 import io.github.eschizoid.kpipe.producer.tracing.Tracer;
 import io.github.eschizoid.kpipe.registry.MessagePipeline;
 import io.github.eschizoid.kpipe.registry.Result;
-import io.github.eschizoid.kpipe.sink.BatchPolicy;
-import io.github.eschizoid.kpipe.sink.BatchSink;
 import java.lang.System.Logger;
 import java.lang.System.Logger.Level;
 import java.nio.channels.ClosedByInterruptException;
@@ -20,14 +17,11 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
-import java.util.function.Function;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.clients.producer.*;
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 
 /// A functional-style Kafka consumer that processes records using a provided function.
 ///
@@ -128,7 +122,8 @@ public class KPipeConsumer implements AutoCloseable {
   /// Composes pause arbitration + backpressure decision + circuit-breaker state machine. The
   /// underlying decision modules ([BackpressureController], [CircuitBreakerController]) remain
   /// public + testable on their own; this controller owns the side-effect choreography and
-  /// dispatches pause transitions through a PauseLifecycleHook that points back at `internalPause` /
+  /// dispatches pause transitions through a PauseLifecycleHook that points back at `internalPause`
+  // /
   /// `internalResume`, and metric events through a HealthMetricsObserver bound to the counters.
   private final ConsumerHealthController health;
 
@@ -178,7 +173,6 @@ public static KPipeConsumerBuilder builder() {
     return new KPipeConsumerBuilder();
   }
 
-
   /// Creates a new KPipeConsumer using the provided builder.
   ///
   /// @param builder the builder containing the consumer configuration
@@ -454,9 +448,11 @@ public void onBatchFailure(final ConsumerRecord<byte[], byte[]> record, final Ex
         otelMetrics.recordProcessingError(record.topic());
         health.recordOutcome(false);
         LOGGER.log(Level.WARNING, () -> "Batch failure for record at offset " + record.offset(), cause);
-        // LOCKSTEP: mirror of the per-record path's DLQ-or-mark block in handleProcessingError —
+        // LOCKSTEP: mirror of the per-record path's DLQ-or-mark block in handleProcessingError
+        // —
         // mark the offset only after a successful DLQ send; a failed send leaves it pending so
-        // the record is reprocessed, never dropped. Deliberately duplicated (the paths differ on
+        // the record is reprocessed, never dropped. Deliberately duplicated (the paths differ
+        // on
         // span handling, retry counts, and circuit-breaker ordering — recordOutcome runs BEFORE
         // this block here, AFTER it on the per-record path). Any change here must be mirrored
         // there; DlqTerminalContractTest asserts both paths cell-for-cell and fails on drift.
@@ -876,6 +872,17 @@ public boolean isRunning() {
     return s == ConsumerState.RUNNING || s == ConsumerState.PAUSED;
   }
 
+  /// Point-in-time health snapshot: running/paused state, active pause sources, circuit-breaker
+  /// state, and in-flight count. This is the source of truth for liveness probes — see
+  /// [HealthSnapshot#healthy()] for why `isRunning()` alone is the wrong probe signal (a
+  /// breaker-tripped consumer is alive but deliberately not consuming).
+  ///
+  /// @return an immutable snapshot of this consumer's health
+  public HealthSnapshot health() {
+    final var sources = health.currentSources().stream().map(Enum::name).collect(Collectors.toUnmodifiableSet());
+    return new HealthSnapshot(isRunning(), health.isPaused(), sources, health.circuitBreakerState(), totalInFlight());
+  }
+
   /// Atomically transitions from any active state (RUNNING or PAUSED) to CLOSING.
   /// Uses a single-read CAS to avoid the double-CAS window.
   ///

lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KafkaOffsetManager.java

@@ -296,6 +296,9 @@ private boolean performCommit(final Map<TopicPartition, OffsetAndMetadata> offse
   ///
   /// @param partition The partition to get state for
   /// @return the partition's offset-tracking state
+  /// **Test observation point.** No production caller — jcstress and property suites assert on
+  /// the commit frontier through this window. Not intended for operational dashboards; wire
+  /// [KPipeConsumer#health()] or `getStatistics()` for those.
   public PartitionState getPartitionState(final TopicPartition partition) {
     return ledger.partitionState(partition, state.get());
   }

lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeCircuitBreakerIntegrationTest.java

@@ -79,6 +79,16 @@ void sustainedFailuresTripBreakerAndPauseConsumer() throws InterruptedException
       "exactly one trip should have been recorded"
     );
 
+    // The health snapshot must tell the truth during the incident: the consumer thread is
+    // alive (running) but the breaker is OPEN, so healthy() is false. isRunning() alone reads
+    // true here - the exact false-positive a liveness probe must not serve.
+    final var snapshot = consumer.health();
+    assertTrue(snapshot.running(), "consumer thread is alive while the breaker is open");
+    assertTrue(snapshot.paused(), "breaker pause must be visible in the snapshot");
+    assertEquals(CircuitBreakerState.OPEN, snapshot.circuitBreaker(), "snapshot must expose the OPEN breaker");
+    assertTrue(snapshot.pauseSources().contains("CIRCUIT_BREAKER"), "pause source must name the breaker");
+    assertTrue(!snapshot.healthy(), "an OPEN breaker must make the consumer unhealthy");
+
     consumer.close();
   }