CB-580: fail a ticket when its member reaches a terminal health state

FleetHealthMonitor now requires a failTarget BiConsumer<String,String>
collaborator (no defaulting overload) and calls it exactly once when a
member transitions into GONE or NEVER_READY, via CB-568's idempotent
target-wide abandon() operation. The reason string names the real
terminal state. failTarget invocation retries up to
MAX_FAIL_TARGET_ATTEMPTS (3) within the same transition if it throws,
and never refires on a later tick where the state is unchanged.

Bridged.java wires messages::abandon as the production failTarget.

bridged/src/main/java/dev/ltms/bridged/Bridged.java

@@ -360,8 +360,10 @@ public final class Bridged {
         var healthScheduler = Executors.newSingleThreadScheduledExecutor(r ->
                 Thread.ofVirtual().name("bridge-health-").unstarted(r));
         if (cfg.health() != null && cfg.health().isEnabled()) {
+            // CB-580: a member found GONE/NEVER_READY must fail whatever ticket is waiting on it,
+            // through the same idempotent target-wide operation CB-516 already uses on release.
             healthMonitor = new FleetHealthMonitor(agents, sessions::roster, messages, healthScheduler,
-                    System::nanoTime, cfg.health().intervalOrDefault());
+                    System::nanoTime, cfg.health().intervalOrDefault(), messages::abandon);
             String coverage = FleetHealthMonitor.coverage(true,
                     cfg.health().notifications() != null && cfg.health().notifications().configured());
             if ("detection-only".equals(coverage)) {

bridged/src/main/java/dev/ltms/bridged/health/FleetHealthMonitor.java

@@ -12,34 +12,50 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
 /** Slow whole-fleet evidence collection. It is deliberately separate from the delivery poller. */
 public final class FleetHealthMonitor {
     private static final Logger log = LoggerFactory.getLogger(FleetHealthMonitor.class);
+
+    /** Bounded attempts to run {@link #failTarget} for one transition. Never retried tick-to-tick (CB-580). */
+    static final int MAX_FAIL_TARGET_ATTEMPTS = 3;
+
     private final AgentControl agents;
     private final Supplier<List<MemberSession>> roster;
     private final MessageService messages;
     private final ScheduledExecutorService scheduler;
     private final LongSupplier clock;
     private final long intervalSeconds;
+    private final BiConsumer<String, String> failTarget;
     private final Map<String, HealthPrior> priors = new HashMap<>();
     private final Map<String, HealthState> states = new HashMap<>();
 
     // These facts need the evidence publishers introduced by later M4 units. They are not negatives.
     private static final boolean NOT_YET_OBSERVED = false;
 
+    /**
+     * @param failTarget CB-568's idempotent target-wide failure operation (e.g. {@code messages::abandon}),
+     *                   invoked once when a member transitions into a terminal health state. Required —
+     *                   there is deliberately no defaulting overload; a caller that does not want the
+     *                   fail-tickets-on-terminal-health behavior must pass an explicit inert value (see
+     *                   {@code TestTurnTokens.inert} / {@code BridgeMcp.CapacitySource.none()} for the pattern).
+     */
     public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
-                              ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds) {
+                              ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds,
+                              BiConsumer<String, String> failTarget) {
         this.agents = agents;
         this.roster = roster;
         this.messages = messages;
         this.scheduler = scheduler;
         this.clock = clock;
         this.intervalSeconds = intervalSeconds;
+        this.failTarget = Objects.requireNonNull(failTarget, "failTarget");
     }
 
     /** Pure per-member decision seam. */
@@ -91,6 +107,34 @@ public final class FleetHealthMonitor {
         } else if (previous != null && fault(previous)) {
             log.info("fleet health member={} recovered state={} previous={}", target, next, previous);
         }
+        // CB-580: a member entering GONE/NEVER_READY must not leave its waiting tickets pending
+        // forever. Fire exactly once per transition — never on a tick where the state is unchanged,
+        // which is what made the rejected commit call abandon() once per tick for as long as a
+        // member stayed terminal.
+        if (terminal(next)) {
+            failTerminalTarget(target, next);
+        }
+    }
+
+    private void failTerminalTarget(String target, HealthState state) {
+        String reason = "fleet health: member reached terminal state " + state.name();
+        RuntimeException last = null;
+        for (int attempt = 1; attempt <= MAX_FAIL_TARGET_ATTEMPTS; attempt++) {
+            try {
+                failTarget.accept(target, reason);
+                return;
+            } catch (RuntimeException error) {
+                last = error;
+                log.warn("fleet health: failTarget attempt {}/{} failed for member={} state={}",
+                        attempt, MAX_FAIL_TARGET_ATTEMPTS, target, state, error);
+            }
+        }
+        log.warn("fleet health: giving up on failTarget for member={} state={} after {} attempts",
+                target, state, MAX_FAIL_TARGET_ATTEMPTS, last);
+    }
+
+    private static boolean terminal(HealthState state) {
+        return state == HealthState.GONE || state == HealthState.NEVER_READY;
     }
 
     private static boolean fault(HealthState state) {