CB-576: prove teardown survives an already-gone worktree

asyncTasksByWaiter correlates an async question by the exact rendezvous
waiter, so the target-keyed set it replaced can no longer decide
anything. Keeping it meant two indexes of the same fact, one of them
ambiguous whenever a target has two accepted tickets.

The race the old test modelled by reflection is gone with it: identity
keys make 'some other task reached this target' unrepresentable, so
there is no longer a wrong task for the question to land on.

Also carries the criterion-1 doc correction, which is identical to
aac29d6 on a different parent.

bridged/bridged.example.yaml

@@ -85,6 +85,16 @@ bind:
 #   backoffMs: 60000
 #   quietNudgeCap: 3
 
+# Fleet health detection is dormant unless enabled. It reads one whole-fleet agent list per tick.
+# It can run without a webhook; bridge_list then reports healthCoverage: detection-only.
+# health:
+#   enabled: true
+#   intervalSeconds: 30              # minimum 15
+#   workingSuspectAfterSeconds: 600  # minimum 300
+#   paneProbeIntervalSeconds: 60     # minimum 60
+#   notifications:
+#     mode: disabled                 # disabled (default) or webhook
+
 # herdr Unix socket. Omit to use the client default
 # (${HERDR_SOCKET_PATH:-~/.config/herdr/herdr.sock}).
 herdrSocket: ~/.config/herdr/herdr.sock

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

@@ -23,6 +23,7 @@ import dev.ltms.bridged.mcp.BridgeMcp;
 import dev.ltms.bridged.mcp.ConnectionIdentity;
 import dev.ltms.bridged.metrics.BridgedMetrics;
 import dev.ltms.bridged.metrics.Metrics;
+import dev.ltms.bridged.health.FleetHealthMonitor;
 import dev.ltms.bridged.mcp.PrimaryRegistry;
 import dev.ltms.bridged.mcp.LsofPeerPidLookup;
 import dev.ltms.bridged.mcp.LsofProcessCwdLookup;
@@ -275,9 +276,9 @@ public final class Bridged {
             }
 
             @Override
-            public void onDelivered(String target) {
-                completion.onDelivered(target);
-                sessions.onDelivered(target);
+            public void onDelivered(String target, dev.ltms.bridged.msg.TurnToken token) {
+                completion.onDelivered(target, token);
+                sessions.onDelivered(target, token);
             }
 
             @Override
@@ -285,6 +286,12 @@ public final class Bridged {
                 completion.onTurnFailed(target);
                 sessions.onTurnFailed(target);
             }
+
+            @Override
+            public void onTurnFailed(String target, String reason) {
+                completion.onTurnFailed(target, reason);
+                sessions.onTurnFailed(target);
+            }
         };
         Injector injector = new Injector(agents, turnListener, deliverableTo(presence, leads),
                 presence::forget);
@@ -348,6 +355,26 @@ public final class Bridged {
         MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
                 pushLoop, metrics);
 
+        // Health is a slow whole-fleet observer. Keep it separate from the 250ms delivery poller.
+        final FleetHealthMonitor healthMonitor;
+        var healthScheduler = Executors.newSingleThreadScheduledExecutor(r ->
+                Thread.ofVirtual().name("bridge-health-").unstarted(r));
+        if (cfg.health() != null && cfg.health().isEnabled()) {
+            healthMonitor = new FleetHealthMonitor(agents, sessions::roster, messages, healthScheduler,
+                    System::nanoTime, cfg.health().intervalOrDefault());
+            String coverage = FleetHealthMonitor.coverage(true,
+                    cfg.health().notifications() != null && cfg.health().notifications().configured());
+            if ("detection-only".equals(coverage)) {
+                log.warn("fleet health: {} (no notification sink configured)", coverage);
+            } else {
+                log.info("fleet health: {}", coverage);
+            }
+            healthMonitor.start();
+        } else {
+            healthMonitor = null;
+            healthScheduler.shutdownNow();
+        }
+
         // CB-520: the reply inbox only consumes for agents this gateway owns. own on acquire,
         // release on teardown. Do this before CB-516 so the inbox is owned before any reply can land.
         sessions.onAcquire(replyInbox::own);
@@ -387,7 +414,12 @@ public final class Bridged {
                 profile -> {
                     var configured = config.get().profiles().get(profile);
                     return configured == null ? null : configured.maxLoad();
-                }, () -> config.get().profiles().keySet(), System::nanoTime));
+                }, () -> config.get().profiles().keySet(), System::nanoTime),
+                new BridgeMcp.HealthCoverageSource(() -> {
+                    var health = config.get().health();
+                    return FleetHealthMonitor.coverage(health != null && health.isEnabled(),
+                            health != null && health.notifications() != null && health.notifications().configured());
+                }));
 
         // CB-559: opt-in config reload. With no `configReload:` block nothing is constructed, so an
         // upgraded daemon behaves exactly as before — the file is read once at boot and never again.
@@ -408,6 +440,7 @@ public final class Bridged {
             messages.close();
             pushLoop.close();
             if (heartbeat != null) heartbeat.close(); // CB-551: stop the idle-lead heartbeat scheduler
+            if (healthMonitor != null) healthMonitor.stop();
             if (configWatcher != null) configWatcher.stop(); // CB-559: stop polling the config file
             mcp.close();
             if (reaper != null) reaper.stop();

bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java

@@ -73,6 +73,7 @@ public record BridgedConfig(
         Primary primary,
         Fleet fleet,
         LeadHeartbeat leadHeartbeat,
+        Health health,
         String placement,
         Auth auth,
         ConfigReload configReload) {
@@ -83,7 +84,16 @@ public record BridgedConfig(
                          Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet,
                          LeadHeartbeat leadHeartbeat, String placement, Auth auth) {
         this(bind, herdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs,
-                spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, placement, auth, null);
+                spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, null);
+    }
+
+    /** Back-compat form before the optional {@code health:} block was added. */
+    public BridgedConfig(Bind bind, String herdrSocket, Map<String, Profile> profiles, Guard guard,
+                         String worktreeRoot, Lifecycle lifecycle, Integer spawnReadyTimeoutMs,
+                         Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet,
+                         LeadHeartbeat leadHeartbeat, String placement, Auth auth, ConfigReload configReload) {
+        this(bind, herdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs,
+                spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, configReload);
     }
 
     /**
@@ -377,6 +387,17 @@ public record BridgedConfig(
                             boolean clearAfterTurn) {
     }
 
+    /** Optional fleet detection. A missing block stays dormant. */
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public record Health(Boolean enabled, Integer intervalSeconds, Integer workingSuspectAfterSeconds,
+                         Integer paneProbeIntervalSeconds, Notifications notifications) {
+        public boolean isEnabled() { return Boolean.TRUE.equals(enabled); }
+        public int intervalOrDefault() { return Math.max(15, intervalSeconds == null ? 30 : intervalSeconds); }
+        public record Notifications(String mode) {
+            public boolean configured() { return "webhook".equalsIgnoreCase(mode); }
+        }
+    }
+
     /**
      * External AMQP broker for durable, cross-restart reply delivery (CB-307 Stage 2). Its mere
      * presence swaps the in-memory {@code ReplyInbox} for the AMQP-backed adapter; absent, bridged
@@ -812,7 +833,7 @@ public record BridgedConfig(
     private static final Set<String> KNOWN_TOP_LEVEL_KEYS = Set.of(
             "bind", "herdrSocket", "profiles", "guard", "worktreeRoot",
             "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs", "broker", "primary", "fleet",
-            "leadHeartbeat", "placement", "auth", "configReload");
+            "leadHeartbeat", "health", "placement", "auth", "configReload");
 
     /** Load and validate config from {@code path}. */
     public static BridgedConfig load(Path path) {
@@ -1082,7 +1103,7 @@ public record BridgedConfig(
         // defaults the fields of a block that IS present. Defaulting it here would start watching
         // the file for every config that never asked to be watched.
         return new BridgedConfig(b, herdrSocket, profiles, g, worktreeRoot, l, timeout, pollMs,
-                broker, primary, f, leadHeartbeat, placementOrDefault, a, configReload);
+                broker, primary, f, leadHeartbeat, health, placementOrDefault, a, configReload);
     }
 
     /**

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

@@ -0,0 +1,107 @@
+package dev.ltms.bridged.health;
+
+import dev.ltms.bridged.herdr.Agent;
+import dev.ltms.bridged.herdr.AgentControl;
+import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.msg.MessageService;
+import dev.ltms.bridged.session.MemberSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+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);
+    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 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;
+
+    public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
+                              ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds) {
+        this.agents = agents;
+        this.roster = roster;
+        this.messages = messages;
+        this.scheduler = scheduler;
+        this.clock = clock;
+        this.intervalSeconds = intervalSeconds;
+    }
+
+    /** Pure per-member decision seam. */
+    static HealthDecision decide(HealthSnapshot snapshot, HealthPrior prior, long nowNanos) {
+        return FleetHealth.decide(snapshot, prior, nowNanos);
+    }
+
+    public void start() { scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS); }
+    public void stop() { scheduler.shutdownNow(); }
+
+    // Package-private so tests can run one tick without waiting.
+    void tick() {
+        try {
+            List<Agent> agentsNow = agents.list(); // Exactly one list call for this complete observation.
+            List<MemberSession> rosterNow = roster.get(); // One in-memory roster snapshot for this tick.
+            Map<String, Agent> live = new HashMap<>();
+            for (Agent agent : agentsNow) live.put(agent.terminalId(), agent);
+            HashSet<String> current = new HashSet<>();
+            for (MemberSession session : rosterNow) {
+                current.add(session.terminalId());
+                Agent agent = live.get(session.terminalId());
+                AgentStatus status = agent == null ? AgentStatus.UNKNOWN : agent.status();
+                boolean accepted = messages.hasAcceptedDelivery(session.terminalId());
+                HealthSnapshot snapshot = new HealthSnapshot(session.state(), status, accepted, NOT_YET_OBSERVED,
+                        messages.hasInboxMessage(session.terminalId()), agent != null, NOT_YET_OBSERVED,
+                        NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED);
+                HealthDecision decision = decide(snapshot, priors.getOrDefault(session.terminalId(), HealthPrior.NONE),
+                        clock.getAsLong());
+                priors.put(session.terminalId(), decision.prior());
+                reportTransition(session.terminalId(), decision.state());
+            }
+            priors.keySet().retainAll(current);
+            states.keySet().retainAll(current);
+        } catch (Throwable error) {
+            // A list failure is health evidence, and must never kill the monitor's only scheduler task.
+            log.warn("fleet health collection failed; will retry next tick", error);
+        } finally {
+            if (!scheduler.isShutdown()) {
+                scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS);
+            }
+        }
+    }
+
+    void reportTransition(String target, HealthState next) {
+        HealthState previous = states.put(target, next);
+        if (previous == next) return;
+        if (fault(next)) {
+            log.warn("fleet health member={} state={} previous={}", target, next, previous);
+        } else if (previous != null && fault(previous)) {
+            log.info("fleet health member={} recovered state={} previous={}", target, next, previous);
+        }
+    }
+
+    private static boolean fault(HealthState state) {
+        return switch (state) {
+            case NEVER_READY, GONE, TURN_BOUNDARY_LOST, ERROR_ON_SCREEN, STALL_SUSPECTED,
+                    MUTE, REPLY_STRANDED, DELEGATION_ORPHANED, CONTROL_LINK_DOWN -> true;
+            default -> false;
+        };
+    }
+
+    public static String coverage(boolean enabled, boolean notificationConfigured) {
+        return !enabled ? "off" : notificationConfigured ? "full" : "detection-only";
+    }
+}

bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java

@@ -2,6 +2,7 @@ package dev.ltms.bridged.inject;
 
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -83,17 +84,17 @@ public final class CompletionResolver implements TurnListener {
     }
 
     @Override
-    public void onDelivered(String target) {
+    public void onDelivered(String target, TurnToken token) {
         // Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn
         // content — what it shows *before* the just-delivered turn produces output — as the staleness
         // reference (CB-115). Done synchronously (like the delivering send itself) so both are in
         // place before this turn's completion can fire.
-        captureBaseline(target);
+        captureBaseline(target, token);
     }
 
     /** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
-    void captureBaseline(String target) {
-        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
+    void captureBaseline(String target, TurnToken token) {
+        CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
         if (waiter == null) {
             inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later
             return;
@@ -137,7 +138,13 @@ public final class CompletionResolver implements TurnListener {
     @Override
     public void onTurnFailed(String target) {
         InFlight turn = inFlight.get(target);
-        Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn));
+        Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn, null));
+    }
+
+    @Override
+    public void onTurnFailed(String target, String reason) {
+        InFlight turn = inFlight.get(target);
+        Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn, reason));
     }
 
     /** Synchronous resolve (the unit-testable core of {@link #onTurnComplete}). */
@@ -192,6 +199,11 @@ public final class CompletionResolver implements TurnListener {
 
     /** Synchronous fail (the unit-testable core of {@link #onTurnFailed}). */
     void fail(String target, InFlight turn) {
+        fail(target, turn, null);
+    }
+
+    /** Synchronous fail with an optional reason supplied by a dropped worker queue. */
+    void fail(String target, InFlight turn, String explicitReason) {
         // A never-delivered readiness failure has no in-flight record but still has a blocked send;
         // fall back to the currently-registered waiter (unambiguous — that send never completed, so
         // no next turn exists to confuse it with).
@@ -201,16 +213,18 @@ public final class CompletionResolver implements TurnListener {
             inFlight.remove(target, turn); // nobody blocked on this worker — nothing to fail
             return;
         }
-        String reason;
-        try {
-            reason = clip(agents.read(target, SCRAPE_SOURCE));
-        } catch (RuntimeException e) {
-            reason = "";
-        }
-        if (reason.isBlank()) {
-            // No screen to scrape — either the worker is stuck (CB-109) or gone (CB-110).
-            reason = "worker did not reply; its turn ended in an unrecoverable state "
-                    + "(worker unreachable or stuck)";
+        String reason = explicitReason;
+        if (reason == null || reason.isBlank()) {
+            try {
+                reason = clip(agents.read(target, SCRAPE_SOURCE));
+            } catch (RuntimeException e) {
+                reason = "";
+            }
+            if (reason.isBlank()) {
+                // No screen to scrape — either the worker is stuck (CB-109) or gone (CB-110).
+                reason = "worker did not reply; its turn ended in an unrecoverable state "
+                        + "(worker unreachable or stuck)";
+            }
         }
         if (rendezvous.resolveFailure(waiter, reason)) {
             inFlight.remove(target, turn);

bridged/src/main/java/dev/ltms/bridged/inject/Injector.java

@@ -2,6 +2,7 @@ package dev.ltms.bridged.inject;
 
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -129,7 +130,7 @@ public final class Injector {
     }
 
     /** A pending message and the future that completes when it has been delivered. */
-    private record Pending(String text, CompletableFuture<Void> delivered) {
+    private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
     }
 
     /** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -159,9 +160,9 @@ public final class Injector {
      * <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
      * target" and "queue the message" and orphan it in a target it just removed.
      */
-    public CompletableFuture<Void> enqueue(String target, String text) {
+    public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
         CompletableFuture<Void> delivered = new CompletableFuture<>();
-        Pending p = new Pending(text, delivered);
+        Pending p = new Pending(text, token, delivered);
         targets.compute(target, (_, existing) -> {
             Target t = (existing != null) ? existing : new Target();
             t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
@@ -356,7 +357,7 @@ public final class Injector {
             } else {
                 // Baseline the pane's pre-turn content so a misattributed completion (no new output)
                 // can't resolve this send with the previous turn's stale answer (CB-115).
-                turnListener.onDelivered(target);
+                turnListener.onDelivered(target, sent.token());
                 sent.delivered().complete(null);
             }
         }
@@ -410,8 +411,8 @@ public final class Injector {
         for (Pending p : pending) {
             p.delivered().completeExceptionally(cause);
         }
-        if (hadDeliveredTurn) {
-            turnListener.onTurnFailed(target);
-        }
+        // A queued send has no in-flight record, while a delivered turn does. CompletionResolver
+        // handles both forms and resolves its waiter at most once.
+        turnListener.onTurnFailed(target, cause.getMessage());
     }
 }

bridged/src/main/java/dev/ltms/bridged/inject/TurnListener.java

@@ -1,5 +1,7 @@
 package dev.ltms.bridged.inject;
 
+import dev.ltms.bridged.msg.TurnToken;
+
 /**
  * Notified when a worker's delegated turn is observed to complete — a confirmed
  * {@code WORKING → IDLE} transition after a delivery. This is the CB-106 completion signal the
@@ -41,6 +43,14 @@ public interface TurnListener {
     default void onTurnFailed(String target) {
     }
 
+    /**
+     * As {@link #onTurnFailed(String)}, carrying the reason a worker became unreachable. The default
+     * keeps existing listeners working while allowing the completion resolver to report a useful cause.
+     */
+    default void onTurnFailed(String target, String reason) {
+        onTurnFailed(target);
+    }
+
     /**
      * A message was just delivered into {@code target}'s pane (CB-115). Fired so the completion
      * resolver can snapshot the pane's pre-turn content: a later {@link #onTurnComplete} whose
@@ -49,7 +59,7 @@ public interface TurnListener {
      * resolve the send with the previous turn's stale answer. A default no-op keeps the interface
      * functional for callers that don't scrape.
      */
-    default void onDelivered(String target) {
+    default void onDelivered(String target, TurnToken token) {
     }
 
     /** No-op default for callers that only need delivery, not completion signalling. */

bridged/src/main/java/dev/ltms/bridged/logging/McpCancelledNotificationFilter.java

@@ -0,0 +1,30 @@
+package dev.ltms.bridged.logging;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.turbo.TurboFilter;
+import ch.qos.logback.core.spi.FilterReply;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.slf4j.Marker;
+
+/** Suppresses only the SDK warning for the normal MCP cancellation notification. */
+public final class McpCancelledNotificationFilter extends TurboFilter {
+
+    static final String LOGGER = "io.modelcontextprotocol.spec.McpStreamableServerSession";
+    static final String UNHANDLED_NOTIFICATION = "No handler registered for notification method: {}";
+
+    @Override
+    public FilterReply decide(Marker marker, Logger logger, Level level, String format, Object[] params,
+                              Throwable throwable) {
+        if (level == Level.WARN
+                && LOGGER.equals(logger.getName())
+                && UNHANDLED_NOTIFICATION.equals(format)
+                && params != null
+                && params.length == 1
+                && params[0] instanceof McpSchema.JSONRPCNotification notification
+                && "notifications/cancelled".equals(notification.method())) {
+            return FilterReply.DENY;
+        }
+        return FilterReply.NEUTRAL;
+    }
+}

bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java

@@ -77,6 +77,7 @@ public final class BridgeMcp {
     private final CallerResolver authz; // CB-501: null → authorization not enforced (legacy)
     private final Metrics metrics;      // CB-502: null → auth failures not counted
     private final CapacitySource capacity;
+    private final HealthCoverageSource healthCoverage;
 
     /** Capacity facts used by {@code bridge_list}; production must supply the placement live count. */
     public record CapacitySource(Function<String, Integer> liveCount, Function<String, Integer> maxLoad,
@@ -86,6 +87,9 @@ public final class BridgeMcp {
         boolean available() { return !configuredProfiles.get().isEmpty(); }
     }
 
+    /** Coverage is supplied by the health wiring, not inferred from a missing dependency. */
+    public record HealthCoverageSource(Supplier<String> value) { }
+
     /**
      * @param callers resolves each call's {@link Principal}; {@code null} disables authorization.
      *                This surface needs its own enforcement: {@code /mcp} is a raw servlet on
@@ -95,8 +99,9 @@ public final class BridgeMcp {
      */
     public BridgeMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
                      ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
-                     CallerResolver callers, Metrics metrics, CapacitySource capacity) {
+                      CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage) {
         this.capacity = capacity;
+        this.healthCoverage = healthCoverage;
         McpJsonMapper json = new JacksonMcpJsonMapperSupplier().get();
         this.transport = HttpServletStreamableServerTransportProvider.builder()
                 .jsonMapper(json)
@@ -210,7 +215,7 @@ public final class BridgeMcp {
                 .toolCall(listTool(), (exchange, _) -> {
                     McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
                     if (denied != null) return denied;
-                    return listFleet(workers, sessions, messages, capacity,
+                    return listFleet(workers, sessions, messages, capacity, healthCoverage,
                             callers == null ? Map.of() : callers.leads(),
                             callerTerminal(exchange));
                 })
@@ -724,11 +729,11 @@ public final class BridgeMcp {
      */
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions,
                                               Map<String, String> leads, String selfTerm) {
-        return listFleet(workers, sessions, null, CapacitySource.none(), leads, selfTerm);
+        return listFleet(workers, sessions, null, CapacitySource.none(), new HealthCoverageSource(() -> "off"), leads, selfTerm);
     }
 
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
-                                              CapacitySource capacity,
+                                               CapacitySource capacity, HealthCoverageSource healthCoverage,
                                               Map<String, String> leads, String selfTerm) {
         try {
             Map<String, Agent> live = workers.list().stream()
@@ -747,6 +752,7 @@ public final class BridgeMcp {
             roster.stream().map(MemberSession::profile).forEach(profiles::add);
             Map<String, Object> result = new LinkedHashMap<>();
             result.put("leads", leadRows); result.put("members", out);
+            result.put("healthCoverage", healthCoverage.value().get());
             if (capacity.available()) result.put("capacity", profiles.stream()
                     .map(profile -> capacityView(profile, capacity.liveCount(), capacity.maxLoad(), roster, messages,
                             capacity.clock().getAsLong())).toList());

bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java

@@ -170,8 +170,9 @@ public final class MessageService {
     private final Metrics metrics; // CB-502: nullable — no registry in unit tests
     private final ConcurrentHashMap<String, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
-    /** The async task that currently owns a target's send lock. */
-    private final ConcurrentHashMap<String, Task> asyncTasksByTarget = new ConcurrentHashMap<>();
+    /** Async task that owns each exact forward rendezvous waiter. */
+    private final ConcurrentHashMap<CompletableFuture<Rendezvous.Resolution>, Task> asyncTasksByWaiter =
+            new ConcurrentHashMap<>();
     /** Async tickets paused on a specific {@code bridge_ask} turn. */
     private final ConcurrentHashMap<String, Task> asyncTasksByTurn = new ConcurrentHashMap<>();
     private final AtomicLong ticketSeq = new AtomicLong();
@@ -300,14 +301,18 @@ public final class MessageService {
      */
     public boolean abandon(String target, String reason) {
         CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
-        if (waiter == null || waiter.isDone()) {
-            return false; // nobody is blocked on this worker — nothing to abandon
+        boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason);
+        boolean asyncFailed = false;
+        for (Task task : tasks.values()) {
+            if (target.equals(task.target) && task.question == null
+                    && task.future.complete(new Reply(Outcome.WORKER_FAILED, reason))) {
+                asyncFailed = true;
+            }
         }
-        boolean failed = rendezvous.resolveFailure(waiter, reason);
         if (failed) {
             log.warn("abandoning the blocked send to {}: {}", target, reason);
         }
-        return failed;
+        return failed || asyncFailed;
     }
 
     /**
@@ -354,6 +359,11 @@ public final class MessageService {
      * never earned. {@code null} disables the hook.
      */
     public Reply send(String target, String content, long timeoutMillis, Runnable onAccepted) {
+        return send(target, content, timeoutMillis, onAccepted, null);
+    }
+
+    /** Run a send, optionally stopping an async task that teardown already failed before acceptance. */
+    private Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, Task task) {
         long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L;
         ReentrantLock lock = sessionLocks.computeIfAbsent(target, _ -> new ReentrantLock());
 
@@ -361,6 +371,9 @@ public final class MessageService {
             return new Reply(Outcome.BUSY, null); // another send held the session the whole window
         }
         try {
+            if (task != null && task.future.isDone()) {
+                return task.future.getNow(null);
+            }
             if (hasAsyncQuestion(target)) {
                 return new Reply(Outcome.BUSY, null); // the worker's current turn is paused for its lead
             }
@@ -372,13 +385,17 @@ public final class MessageService {
             // failed send leaves no stale waiter behind.
             CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
             try {
+                if (task != null) {
+                    asyncTasksByWaiter.put(reply, task);
+                }
+                TurnToken token = new TurnToken(target, reply);
                 // The send has won the lock; the accepted-delivery hook records delegator ownership
                 // here (CB-548). It runs BEFORE enqueue so a throwing hook — onAccepted is now a
                 // public callback — fails the send without queuing a message that would orphan.
                 if (onAccepted != null) {
                     onAccepted.run();
                 }
-                CompletableFuture<Void> delivered = injector.enqueue(target, content);
+                CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
                 try {
                     Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
                     return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
@@ -395,6 +412,7 @@ public final class MessageService {
                     throw new IllegalStateException("interrupted awaiting reply from " + target, e);
                 }
             } finally {
+                asyncTasksByWaiter.remove(reply);
                 rendezvous.close(target, reply);
             }
         } finally {
@@ -420,11 +438,15 @@ public final class MessageService {
         if (ticket.fresh()) {
             // Register the reverse waiter first, then surface the question — so the answer, which can
             // arrive the instant the primary reacts, always finds an open waiter to resolve.
+            CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(workerSession);
+            Task task = markAsyncQuestion(waiter, question, ticket.turnId());
             if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
+                if (task != null) {
+                    clearAsyncQuestion(ticket.turnId(), true);
+                }
                 rendezvous.closeAsk(ticket.turnId());
                 return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
             }
-            markAsyncQuestion(workerSession, question, ticket.turnId());
         }
         try {
             String answer = ticket.answer().get(timeoutMillis, TimeUnit.MILLISECONDS);
@@ -523,21 +545,15 @@ public final class MessageService {
         tasks.put(ticket, task);
         asyncExecutor.submit(() -> {
             try {
-                Runnable trackingAccepted = () -> {
-                    if (onAccepted != null) {
-                        onAccepted.run();
-                    }
-                    asyncTasksByTarget.put(target, task);
-                };
-                Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted);
+                Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task);
                 if (result.outcome() == Outcome.QUESTION) {
-                    asyncTasksByTarget.remove(target, task);
+                    // Keep the accepted owner until answer() finishes it. markAsyncQuestion may run
+                    // just after resolveQuestion wakes this thread.
                 } else {
                     finishAsyncTask(task, result);
                 }
             } catch (Throwable t) {
                 task.future.completeExceptionally(t);
-                asyncTasksByTarget.remove(target, task);
             }
         });
         pruneTerminalTickets();
@@ -599,13 +615,14 @@ public final class MessageService {
     }
 
     /** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */
-    private void markAsyncQuestion(String target, String text, String turnId) {
-        Task task = asyncTasksByTarget.get(target);
+    private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
+        Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
         if (task != null) {
             task.question = new Reply(Outcome.QUESTION, text, turnId);
             task.turnId = turnId;
             asyncTasksByTurn.put(turnId, task);
         }
+        return task;
     }
 
     /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
@@ -623,7 +640,6 @@ public final class MessageService {
     /** Complete and detach an async ticket after its worker's actual terminal reply. */
     private void finishAsyncTask(Task task, Reply result) {
         task.future.complete(result);
-        asyncTasksByTarget.remove(task.target, task);
         if (task.turnId != null) {
             asyncTasksByTurn.remove(task.turnId, task);
         }

bridged/src/main/java/dev/ltms/bridged/msg/TurnToken.java

@@ -0,0 +1,20 @@
+package dev.ltms.bridged.msg;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Identity for one accepted send. The session turn is deliberately absent: CompletionResolver's
+ * delivery callback runs before SessionManager.onDelivered, so binding it needs a later ordering design.
+ */
+public final class TurnToken {
+    private final String target;
+    private final CompletableFuture<Rendezvous.Resolution> waiter;
+
+    public TurnToken(String target, CompletableFuture<Rendezvous.Resolution> waiter) {
+        this.target = target;
+        this.waiter = waiter;
+    }
+
+    public String target() { return target; }
+    public CompletableFuture<Rendezvous.Resolution> waiter() { return waiter; }
+}

bridged/src/main/java/dev/ltms/bridged/session/GitWorktrees.java

@@ -167,6 +167,22 @@ public final class GitWorktrees implements Worktrees {
         exec("git", "-C", repoRoot, "worktree", "remove", "--force", worktreePath);
     }
 
+    @Override
+    public boolean hasUncommitted(String worktreePath) {
+        // A worktree that is already gone holds no work to lose, and it must not break teardown:
+        // git -C <missing-dir> status exits non-zero and would throw where release() is mid-way
+        // through stopping a pane. Mirror remove()'s already-gone tolerance by treating it as clean.
+        Path p = Path.of(worktreePath);
+        if (!Files.exists(p)) {
+            log.debug("worktree {} already gone — nothing can be uncommitted", worktreePath);
+            return false;
+        }
+        // No --untracked-files=no: the exact shape of the work lost in CB-576 was a new file
+        // that was never added, so an untracked-only worktree is still dirty.
+        String out = exec("git", "-C", worktreePath, "status", "--porcelain");
+        return !out.isBlank();
+    }
+
     @Override
     public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
         if (overlay == null || overlay.isEmpty()) {

bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java

@@ -4,6 +4,7 @@ import dev.ltms.bridged.auth.MemberLifecycle;
 import dev.ltms.bridged.herdr.Agent;
 import dev.ltms.bridged.inject.TurnListener;
 import dev.ltms.bridged.inject.MemberPresence;
+import dev.ltms.bridged.msg.TurnToken;
 import dev.ltms.bridged.peer.MemberRole;
 import dev.ltms.bridged.peer.PeerHandle;
 import dev.ltms.bridged.peer.PeerLauncher;
@@ -200,6 +201,16 @@ public final class SessionManager implements TurnListener {
                     removed.paneId(), removed.terminalId(), removed.state(), cause);
             if (preserveWorktree && removed.worktree() != null) {
                 logPreservedForShutdown(removed);
+            } else if (removed.worktree() != null && worktrees.hasUncommitted(removed.worktree())) {
+                // CB-576: a release that would otherwise remove the worktree finds it holding
+                // uncommitted work the bridge cannot see. A worker that ends a turn without
+                // committing (normally because it stopped to ask a question or refused the turn)
+                // has its only copy of that work in the worktree. Remove would --force-delete it,
+                // so preserve the directory and tell an operator where to find it.
+                preserveWorktree = true;
+                log.warn("release {} preserves dirty worktree {} for pane={} terminal={}: "
+                        + "the worktree holds uncommitted changes that --force remove would destroy",
+                        cause, removed.worktree(), removed.paneId(), removed.terminalId());
             }
             // CB-516: a send still waiting on this worker can never be answered now. Tell the
             // listener BEFORE the pane is torn down, so a blocked caller fails fast with a real
@@ -425,7 +436,7 @@ public final class SessionManager implements TurnListener {
      * can be re-delivered for multi-turn reuse until it is released.
      */
     @Override
-    public void onDelivered(String target) {
+    public void onDelivered(String target, TurnToken token) {
         MemberSession current = findByTerminal(target);
         if (current == null) return;
         if (current.state() != MemberSession.State.READY && current.state() != MemberSession.State.DONE) {

bridged/src/main/java/dev/ltms/bridged/session/Worktrees.java

@@ -10,6 +10,18 @@ public interface Worktrees {
     /** git -C <repoRoot> worktree remove --force <path>. Idempotent (already-gone tolerated). */
     void remove(String repoRoot, String worktreePath);
 
+    /**
+     * True when the worktree holds uncommitted changes the bridge cannot see: tracked
+     * modifications, staged files, or untracked files. {@code git status --porcelain} is the
+     * test; an empty result means clean. Callers use this to decide whether removing the
+     * worktree would silently destroy a worker's only copy of its work.
+     *
+     * <p>An already-gone worktree is reported as clean (no throw), matching {@link #remove}'s
+     * idempotent contract: a path that does not exist holds no work to lose, and must not break
+     * a teardown that is mid-way through stopping the pane.
+     */
+    boolean hasUncommitted(String worktreePath);
+
     /** Copy each existing overlay path repoRoot→worktree; mark tracked ones --skip-worktree. */
     void overlayParity(String repoRoot, String worktreePath, List<String> overlay);
 

bridged/src/main/resources/logback.xml

@@ -1,4 +1,13 @@
 <configuration>
+    <!--
+      CB-575: MCP SDK 2.0.0 has no public notification registration API. It registers only
+      notifications/initialized and notifications/roots/list_changed, so other client notifications
+      still warn when unhandled. Clients may legitimately send notifications/cancelled; suppress only
+      that SDK WARN because it would devalue the action-needed WARN level used by M4 fleet health.
+      If a later SDK handles cancellation, this filter simply stops matching and can be removed.
+    -->
+    <turboFilter class="dev.ltms.bridged.logging.McpCancelledNotificationFilter"/>
+
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
             <pattern>%d{HH:mm:ss.SSS} %-5level [%thread] %logger{28} - %msg%n</pattern>

bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java

@@ -0,0 +1,81 @@
+package dev.ltms.bridged.health;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.ltms.bridged.herdr.AgentControl;
+import dev.ltms.bridged.herdr.FakeHerdr;
+import dev.ltms.bridged.inject.Injector;
+import dev.ltms.bridged.msg.InMemoryReplyInbox;
+import dev.ltms.bridged.msg.MessageService;
+import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.session.SessionManager;
+import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.herdr.WorkspaceControl;
+import dev.ltms.bridged.guard.SubscriptionGuard;
+import dev.ltms.bridged.config.BridgedConfig;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class FleetHealthMonitorTest {
+    @Test void oneTickUsesOneFleetListForAnyRosterSize() {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        BridgedConfig.Profile profile = new BridgedConfig.Profile("test", "http://test:1", null,
+                null, null, null, null, null, null, null, null, null);
+        ClaudeCodeLauncher launcher = new ClaudeCodeLauncher(agents, new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("test")), Map.of("test", profile), "test", _ -> "token");
+        SessionManager sessions = new SessionManager(launcher);
+        sessions.acquire("test", null, null, null);
+        sessions.acquire("test", null, null, null);
+        herdr.calls.clear();
+        MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox());
+        var scheduler = Executors.newSingleThreadScheduledExecutor();
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, sessions::roster, messages, scheduler, () -> 1, 60);
+        monitor.tick();
+        monitor.stop();
+        assertEquals(1, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
+    }
+
+    @Test void failedTickDoesNotStopTheNextTick() {
+        FakeHerdr herdr = new FakeHerdr().healthy(false);
+        AgentControl agents = new AgentControl(herdr);
+        var scheduler = Executors.newSingleThreadScheduledExecutor();
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
+                new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
+                scheduler, () -> 1, 60);
+        monitor.tick();
+        herdr.healthy(true);
+        monitor.tick();
+        monitor.stop();
+        assertEquals(2, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
+    }
+
+    @Test void faultTransitionLogsOnlyOnceUntilItChanges() {
+        Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            FakeHerdr herdr = new FakeHerdr();
+            AgentControl agents = new AgentControl(herdr);
+            var scheduler = Executors.newSingleThreadScheduledExecutor();
+            FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
+                    new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
+                    scheduler, () -> 1, 60);
+            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
+            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
+            monitor.stop();
+            assertEquals(1, appender.list.stream().filter(event -> event.getFormattedMessage()
+                    .contains("member=term_a state=TURN_BOUNDARY_LOST")).count());
+        } finally {
+            logger.detachAppender(appender);
+        }
+    }
+}

bridged/src/test/java/dev/ltms/bridged/inject/CompletionResolverTest.java

@@ -7,6 +7,8 @@ import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.msg.TestTurnTokens;
+import dev.ltms.bridged.msg.TurnToken;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
 
@@ -47,7 +49,7 @@ class CompletionResolverTest {
         Rendezvous rendezvous = new Rendezvous(); // no waiter opened
         CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
 
-        resolver.captureBaseline("term_a"); // no send to attribute a later completion to
+        resolver.captureBaseline("term_a", TestTurnTokens.inert("term_a")); // no send to attribute a later completion to
 
         assertFalse(herdr.called("agent.read"),
                 "with no waiting send there is no turn to baseline — skip the scrape");
@@ -194,7 +196,7 @@ class CompletionResolverTest {
         Rendezvous rendezvous = new Rendezvous();
         CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
         var waiter = rendezvous.open("term_a");
-        resolver.captureBaseline("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
         herdr.readText("⏺ answer that /clear would erase\n❯ ");
 
         resolver.resolveBeforePostAction("term_a");
@@ -217,7 +219,7 @@ class CompletionResolverTest {
         CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
 
         var waiter = rendezvous.open("term_a");        // a send is blocked on this turn
-        resolver.captureBaseline("term_a");            // baseline is the clipped >cap block
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiter)); // baseline is the clipped >cap block
         var turn = resolver.inFlight("term_a");
         assertEquals(CompletionResolver.MAX_SCRAPE_CHARS, turn.baseline().length(),
                 "the delivery baseline is clipped to the same cap resolve() applies to the tail");

bridged/src/test/java/dev/ltms/bridged/inject/InjectorTest.java

@@ -8,6 +8,7 @@ import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.HerdrException;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
 
@@ -47,7 +48,7 @@ class InjectorTest {
 
     @Test
     void deliversWhenIdle() {
-        CompletableFuture<Void> f = injector.enqueue(T, "hello");
+        CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
         assertFalse(f.isDone(), "not delivered until an injectable status arrives");
         injector.onStatus(T, AgentStatus.IDLE);
         assertTrue(f.isDone());
@@ -59,7 +60,7 @@ class InjectorTest {
         // CB-113: idle alone is not enough — hold until the worker's MCP is connected (ready).
         java.util.Set<String> ready = new java.util.HashSet<>();
         Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE); // idle but not yet available → held out of the boot window
         assertEquals(List.of(), sent(), "must not deliver into a not-yet-available worker");
@@ -79,7 +80,7 @@ class InjectorTest {
     void resubmitsEnterWhenADeliveredMessageIsNotPickedUp() {
         // CB-113: the Enter at delivery can race the paste; while the worker stays idle (not picked
         // up), the injector re-nudges Enter so the pending paste submits.
-        injector.enqueue(T, "task");
+        injector.enqueue(T, "task", TestTurnTokens.inert(T));
         injector.onStatus(T, AgentStatus.IDLE); // deliver: paste + one Enter
         long afterDeliver = enterKeystrokes();
 
@@ -95,7 +96,7 @@ class InjectorTest {
 
     @Test
     void holdsWhileWorkingThenDeliversOnIdle() {
-        injector.enqueue(T, "later");
+        injector.enqueue(T, "later", TestTurnTokens.inert(T));
         injector.onStatus(T, AgentStatus.WORKING);
         assertEquals(List.of(), sent(), "must not inject mid-turn");
         injector.onStatus(T, AgentStatus.IDLE);
@@ -104,7 +105,7 @@ class InjectorTest {
 
     @Test
     void blockedIsInjectableButUnknownIsNot() {
-        injector.enqueue(T, "answer");
+        injector.enqueue(T, "answer", TestTurnTokens.inert(T));
         injector.onStatus(T, AgentStatus.UNKNOWN);
         assertEquals(List.of(), sent(), "unknown status is not safe to inject");
         injector.onStatus(T, AgentStatus.BLOCKED);
@@ -113,8 +114,8 @@ class InjectorTest {
 
     @Test
     void twoRapidDeliveriesNeverInterleave() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));
 
         // First idle window delivers only m1, even if idle is observed twice before pickup.
         injector.onStatus(T, AgentStatus.IDLE);
@@ -129,8 +130,8 @@ class InjectorTest {
 
     @Test
     void transientUnknownDoesNotReleaseThePickupLatch() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));
         injector.onStatus(T, AgentStatus.IDLE); // m1 sent, awaiting pickup
         assertEquals(List.of("m1"), sent());
 
@@ -145,8 +146,8 @@ class InjectorTest {
 
     @Test
     void missedPickupEdgeIsReleasedByGraceSoTheQueueNeverWedges() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));
         injector.onStatus(T, AgentStatus.IDLE); // m1 sent
         assertEquals(List.of("m1"), sent());
 
@@ -158,9 +159,9 @@ class InjectorTest {
 
     @Test
     void fifoOrderAcrossManyTurns() {
-        injector.enqueue(T, "a");
-        injector.enqueue(T, "b");
-        injector.enqueue(T, "c");
+        injector.enqueue(T, "a", TestTurnTokens.inert(T));
+        injector.enqueue(T, "b", TestTurnTokens.inert(T));
+        injector.enqueue(T, "c", TestTurnTokens.inert(T));
         for (int i = 0; i < 3; i++) {
             injector.onStatus(T, AgentStatus.IDLE);   // deliver one
             injector.onStatus(T, AgentStatus.WORKING); // pickup
@@ -172,7 +173,7 @@ class InjectorTest {
     @Test
     void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
         assertTrue(injector.activeTargets().isEmpty());
-        injector.enqueue(T, "x");
+        injector.enqueue(T, "x", TestTurnTokens.inert(T));
         assertEquals(Set.of(T), injector.activeTargets(), "active while a message is queued");
 
         injector.onStatus(T, AgentStatus.IDLE); // delivers; awaiting pickup
@@ -191,7 +192,7 @@ class InjectorTest {
     void firesTurnCompleteOnAConfirmedWorkingThenIdle() {
         List<String> completed = new ArrayList<>();
         Injector inj = new Injector(new AgentControl(herdr), completed::add);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE);    // deliver
         inj.onStatus(T, AgentStatus.WORKING); // pickup + turn running
@@ -226,8 +227,8 @@ class InjectorTest {
         }
         ResetListener listener = new ResetListener();
         Injector inj = new Injector(agents, listener);
-        inj.enqueue(T, "first");
-        inj.enqueue(T, "second");
+        inj.enqueue(T, "first", TestTurnTokens.inert(T));
+        inj.enqueue(T, "second", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE);    // first delegation
         inj.onStatus(T, AgentStatus.WORKING);
@@ -246,7 +247,7 @@ class InjectorTest {
     void doesNotSynthesizeCompletionFromAnUnconfirmedTurn() {
         List<String> completed = new ArrayList<>();
         Injector inj = new Injector(new AgentControl(herdr), completed::add);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         // Deliver, then only ever idle — a `working` sample is never seen. The pickup grace unwedges
         // the queue but must NOT invent a completion: without a sampled turn there is no trustworthy
@@ -259,6 +260,7 @@ class InjectorTest {
     private static final class Captor implements TurnListener {
         final List<String> completed = new ArrayList<>();
         final List<String> failed = new ArrayList<>();
+        final List<String> failureReasons = new ArrayList<>();
 
         @Override
         public void onTurnComplete(String target) {
@@ -269,6 +271,12 @@ class InjectorTest {
         public void onTurnFailed(String target) {
             failed.add(target);
         }
+
+        @Override
+        public void onTurnFailed(String target, String reason) {
+            failed.add(target);
+            failureReasons.add(reason);
+        }
     }
 
     // ~30s of unknown at the 250ms prod poll interval; enough onStatus samples to trip the stall.
@@ -278,7 +286,7 @@ class InjectorTest {
     void failsAnOutstandingDelegationWhoseWorkerWedgesInUnknown() {
         Captor cap = new Captor();
         Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE);    // deliver
         inj.onStatus(T, AgentStatus.WORKING); // worker starts the turn
@@ -293,7 +301,7 @@ class InjectorTest {
     void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
         Captor cap = new Captor();
         Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE);    // deliver
         inj.onStatus(T, AgentStatus.WORKING); // confirmed turn
@@ -308,7 +316,7 @@ class InjectorTest {
     void sendFailureDropsMessageAndFailsItsFuture() {
         FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
         Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom");
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
 
         inj.onStatus(T, AgentStatus.IDLE);
         assertTrue(f.isCompletedExceptionally());
@@ -317,18 +325,35 @@ class InjectorTest {
 
     @Test
     void dropFailsPendingWaiters() {
-        CompletableFuture<Void> f = injector.enqueue(T, "orphan");
+        CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
         injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
         assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
     }
 
+    @Test
+    void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
+        Captor cap = new Captor();
+        Injector inj = new Injector(new AgentControl(herdr), cap);
+        CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
+        CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));
+
+        inj.onStatus(T, AgentStatus.IDLE);    // deliver the first message
+        inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
+        inj.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
+
+        assertEquals(List.of(T), cap.failed, "drop signals one turn failure for both affected states");
+        assertEquals(List.of("agent target sol not found"), cap.failureReasons);
+        assertTrue(delivered.isDone(), "the delivered future has already completed");
+        assertTrue(queued.isCompletedExceptionally(), "the queued future fails with the drop cause");
+    }
+
     @Test
     void dropFailsTheTurnOfADeliveredMessageWhenTheWorkerVanishes() {
         // CB-110: the message was delivered (no longer queued), so failing queued waiters alone would
         // leave its send hanging. A vanished worker must fail that in-flight turn too.
         Captor cap = new Captor();
         Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
         inj.onStatus(T, AgentStatus.IDLE);    // deliver
         inj.onStatus(T, AgentStatus.WORKING); // turn running
 
@@ -343,7 +368,7 @@ class InjectorTest {
         // by an off-sub worker's review of CB-110, delegated through the bridge.)
         Captor cap = new Captor();
         Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
         inj.onStatus(T, AgentStatus.IDLE); // deliver; pickup never confirmed
 
         inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
@@ -362,7 +387,7 @@ class InjectorTest {
         Captor cap = new Captor();
         List<String> forgotten = new ArrayList<>();
         Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
-        CompletableFuture<Void> f = inj.enqueue(T, "task");
+        CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
 
@@ -390,7 +415,7 @@ class InjectorTest {
         try {
             Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> false, _ -> {
             });
-            inj.enqueue(T, "task");
+            inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
             for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
 
@@ -415,7 +440,7 @@ class InjectorTest {
         Set<String> ready = new java.util.HashSet<>();
         Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains, _ -> {
         });
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
 
         for (int i = 0; i < 100; i++) inj.onStatus(T, AgentStatus.IDLE); // still booting, well under grace
         assertEquals(List.of(), sent());
@@ -431,7 +456,7 @@ class InjectorTest {
         // linger past the worker's life (MemberPresence.forget had no caller before this).
         List<String> forgotten = new ArrayList<>();
         Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, forgotten::add);
-        inj.enqueue(T, "orphan");
+        inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
         inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
         assertEquals(List.of(T), forgotten, "drop clears the gone worker's presence");
     }
@@ -452,7 +477,7 @@ class InjectorTest {
         try {
             Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, _ -> {
             });
-            inj.enqueue(T, "orphan");
+            inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
             inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
 
             String warn = appender.list.stream()
@@ -476,7 +501,7 @@ class InjectorTest {
         StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
         poller.start();
         try {
-            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller");
+            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
             delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
         } finally {
             poller.stop();
@@ -495,7 +520,7 @@ class InjectorTest {
     void deliveredFutureCarriesSendFailure() {
         FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
         Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom");
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
         inj.onStatus(T, AgentStatus.IDLE);
         ExecutionException ex = assertThrows(ExecutionException.class, f::get);
         assertInstanceOf(HerdrException.class, ex.getCause());

bridged/src/test/java/dev/ltms/bridged/logging/McpCancelledNotificationFilterTest.java

@@ -0,0 +1,37 @@
+package dev.ltms.bridged.logging;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class McpCancelledNotificationFilterTest {
+
+    @Test
+    void suppressesOnlyTheCancelledNotificationWarning() {
+        Logger logger = (Logger) LoggerFactory.getLogger(McpCancelledNotificationFilter.LOGGER);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            logger.warn(McpCancelledNotificationFilter.UNHANDLED_NOTIFICATION,
+                    new McpSchema.JSONRPCNotification("notifications/cancelled", Map.of("requestId", 7)));
+            logger.warn(McpCancelledNotificationFilter.UNHANDLED_NOTIFICATION,
+                    new McpSchema.JSONRPCNotification("notifications/progress", Map.of("progress", 1)));
+
+            assertEquals(1, appender.list.size());
+            assertEquals(Level.WARN, appender.list.getFirst().getLevel());
+            assertTrue(appender.list.getFirst().getFormattedMessage().contains("notifications/progress"));
+        } finally {
+            logger.detachAppender(appender);
+        }
+    }
+}

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java

@@ -72,7 +72,7 @@ class BridgeMcpAuthzTest {
                 new PrimaryRegistry(null),
                 enforce ? CallerResolver.withLeadsAndMembers(identity, false, null,
                 Map::of, new MemberRegistry(null)) : null,
-                metrics, BridgeMcp.CapacitySource.none());
+                metrics, BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"));
         return mcp;
     }
 

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java

@@ -474,7 +474,7 @@ class BridgeMcpTest {
         sessions.acquire("ltms-local", null, null, null);
         McpSchema.CallToolResult res = BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 sessions, null, new BridgeMcp.CapacitySource(profile -> 2, profile -> 2,
-                () -> Set.of("ltms-local"), () -> 0), Map.of(), "");
+                () -> Set.of("ltms-local"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), "");
         String out = textOf(res);
         assertTrue(out.contains("\"maxLoad\":2"), out);
         assertTrue(out.contains("\"live\":2"), out);
@@ -487,7 +487,7 @@ class BridgeMcpTest {
         SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
         String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 sessions, null, new BridgeMcp.CapacitySource(profile -> 0, profile -> 2,
-                () -> Set.of("terra"), () -> 0), Map.of(), ""));
+                () -> Set.of("terra"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
         assertTrue(out.contains("\"profile\":\"terra\""), out);
         assertTrue(out.contains("\"live\":0"), out);
         assertTrue(out.contains("\"free\":2"), out);
@@ -499,7 +499,7 @@ class BridgeMcpTest {
         FakeHerdr h = new FakeHerdr();
         String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw"))), null,
-                BridgeMcp.CapacitySource.none(), Map.of(), ""));
+                BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
         assertFalse(out.contains("\"capacity\":"), out);
     }
 

bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java

@@ -120,6 +120,36 @@ class MessageServiceTest {
         assertFalse(reply.completed());
     }
 
+    @Test
+    void droppedQueuedAndDeliveredTurnsExposeTheRealCauseExactlyOnce() throws Exception {
+        CompletableFuture<MessageService.Reply> first = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // first delivery
+        injector.onStatus(T, AgentStatus.WORKING); // first turn in flight
+
+        CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
+        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
+        injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
+
+        MessageService.Reply reply = first.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.WORKER_FAILED, reply.outcome());
+        assertEquals("agent target sol not found", reply.text());
+        assertTrue(queued.isCompletedExceptionally(), "the queued delivery future also fails");
+        assertFalse(rendezvous.resolveFailure(waiter, "second failure"), "the waiter fails exactly once");
+    }
+
+    @Test
+    void noDropReasonKeepsTheExistingFallbackText() throws Exception {
+        herdr.readText("");
+        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.open(T);
+
+        completion.onTurnFailed(T);
+
+        Rendezvous.Resolution resolution = waiter.get(2, TimeUnit.SECONDS);
+        assertEquals("worker did not reply; its turn ended in an unrecoverable state "
+                + "(worker unreachable or stuck)", resolution.text());
+    }
+
     // --- bridge_ask reverse rendezvous (CB-205) ------------------------------------------------
 
     @Test
@@ -609,4 +639,94 @@ class MessageServiceTest {
         assertTrue(view.detail() != null && view.detail().contains("released"),
                 "and the detail says why, rather than 'worker unknown'");
     }
+
+    @Test
+    void abandonFailsEveryPendingAsyncTicketForTheReleasedTarget() throws Exception {
+        String first = messages.sendAsync(T, "first task");
+        awaitWaiting(); // first task owns the target lock and rendezvous waiter
+        String second = messages.sendAsync(T, "second task"); // parked on the same lock, not yet queued
+        String third = messages.sendAsync(T, "third task"); // a second queued ticket proves the full sweep
+
+        assertTrue(messages.abandon(T, "agent target term_a not found"));
+
+        assertFailedTicket(first, "agent target term_a not found");
+        assertFailedTicket(second, "agent target term_a not found");
+        assertFailedTicket(third, "agent target term_a not found");
+    }
+
+    @Test
+    void abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+
+        assertFalse(messages.abandon(T, "agent target term_a not found"),
+                "an asking ticket is an active turn, not a pending send to sweep");
+        assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase());
+
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
+    }
+
+    @Test
+    void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        assertEquals(MessageService.AskOutcome.TIMED_OUT,
+                messages.ask(T, "which config?", 200).outcome());
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
+                "only the question wait ended; the delegated turn may still finish");
+
+        String next = messages.sendAsync(T, "next task");
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
+    }
+
+    @Test
+    void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
+        String first = messages.sendAsync(T, "first task");
+        awaitWaiting();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        assertEquals(MessageService.Phase.ASKING, awaitTicketPhase(first, MessageService.Phase.ASKING).phase());
+
+        MessageService.TaskView asking = messages.poll(first);
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
+    }
+
+    private void assertFailedTicket(String ticket, String reason) throws Exception {
+        MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.FAILED);
+        assertEquals(reason, view.detail());
+    }
+
+    private MessageService.TaskView awaitTicketPhase(String ticket, MessageService.Phase phase) throws Exception {
+        long deadline = System.currentTimeMillis() + 3000;
+        MessageService.TaskView view;
+        do {
+            view = messages.poll(ticket);
+            if (view.phase() == phase) {
+                return view;
+            }
+            Thread.sleep(5);
+        } while (System.currentTimeMillis() < deadline);
+        assertEquals(phase, view.phase());
+        return view;
+    }
 }

bridged/src/test/java/dev/ltms/bridged/msg/TestTurnTokens.java

@@ -0,0 +1,20 @@
+package dev.ltms.bridged.msg;
+
+/**
+ * Explicit unbound tokens for tests that exercise delivery without an accepted send.
+ *
+ * <p>The waiter is {@code null} on purpose. "No accepted send" is an <em>absence</em>, and a helper
+ * that handed back a fresh {@code CompletableFuture} would invent one — which is how the first
+ * version of this class turned {@code captureBaselineSkipsTheReadWhenNoSendIsWaiting} red: the
+ * resolver saw a non-null waiter, decided a turn was in flight, and scraped a pane that no send was
+ * blocked on. An inert value must omit the fact, never fabricate it.
+ */
+public final class TestTurnTokens {
+    private TestTurnTokens() {
+    }
+
+    /** A token for a delivery that no send is waiting on: it authorises nothing. */
+    public static TurnToken inert(String target) {
+        return new TurnToken(target, null);
+    }
+}

bridged/src/test/java/dev/ltms/bridged/session/FakeWorktrees.java

@@ -29,8 +29,11 @@ public final class FakeWorktrees implements Worktrees {
     private final Set<String> existingPaths = ConcurrentHashMap.newKeySet();
     private final Set<String> trackedPaths = ConcurrentHashMap.newKeySet();
     private volatile RuntimeException addFailure;
+    private volatile boolean dirty = false;
     private volatile String repoRoot = "/repo";
     private volatile String prefix = "/worktrees";
+    /** Worktree paths that currently exist, mirroring real {@code Files.exists} for the gone case. */
+    private final Set<String> worktreePaths = ConcurrentHashMap.newKeySet();
 
     public FakeWorktrees withRepoRoot(String root) {
         this.repoRoot = root;
@@ -61,6 +64,12 @@ public final class FakeWorktrees implements Worktrees {
         return this;
     }
 
+    /** Mark the worktree dirty so {@link #hasUncommitted} reports true (simulates uncommitted work). */
+    public FakeWorktrees withDirty(boolean dirty) {
+        this.dirty = dirty;
+        return this;
+    }
+
     @Override
     public String add(String repoRoot, String branch, String baseRef) {
         addCalls.add(new AddCall(repoRoot, branch, baseRef));
@@ -69,7 +78,15 @@ public final class FakeWorktrees implements Worktrees {
         }
         // The branch already carries a unique nonce, so the derived path is distinct per acquire
         // without an extra counter — keep it a pure function of the branch the test can predict.
-        return prefix + "/" + branch.replace('/', '_');
+        String path = prefix + "/" + branch.replace('/', '_');
+        worktreePaths.add(path);
+        return path;
+    }
+
+    /** Model an operator / {@code git worktree prune} removing the worktree before release. */
+    public FakeWorktrees markGone(String worktreePath) {
+        worktreePaths.remove(worktreePath);
+        return this;
     }
 
     @Override
@@ -77,6 +94,16 @@ public final class FakeWorktrees implements Worktrees {
         removeCalls.add(new RemoveCall(repoRoot, worktreePath));
     }
 
+    @Override
+    public boolean hasUncommitted(String worktreePath) {
+        // A path that does not exist (never added, or marked gone) is reported clean, mirroring
+        // GitWorktrees' already-gone guard — never an error, so teardown still completes.
+        if (!worktreePaths.contains(worktreePath)) {
+            return false;
+        }
+        return dirty;
+    }
+
     @Override
     public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
         List<String> copied = new java.util.ArrayList<>();

bridged/src/test/java/dev/ltms/bridged/session/GitWorktreesTest.java

@@ -175,6 +175,46 @@ class GitWorktreesTest {
         assertTrue(Files.exists(Path.of(wt).resolve(".mcp.json")), ".mcp.json stub was dropped");
     }
 
+    /**
+     * CB-576. {@code hasUncommitted} must treat a freshly-provisioned worktree as clean, but a
+     * worktree holding a brand-new, never-added file as dirty. The untracked-file-only shape is
+     * exactly the work lost in the incident — a worker's draft that compiled but was never
+     * committed because it stopped to ask its lead a question.
+     */
+    @Test
+    void anUntrackedOnlyWorktreeCountsAsDirty(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
+        String wt = gitWorktrees.add(repo.toString(), "cb-576-u", "HEAD");
+
+        assertFalse(gitWorktrees.hasUncommitted(wt),
+                "a freshly provisioned worktree must read as clean");
+
+        Files.writeString(Path.of(wt).resolve("brand-new.txt"), "draft that was never added\n");
+
+        assertTrue(gitWorktrees.hasUncommitted(wt),
+                "an untracked-only file must count as dirty");
+
+        Files.writeString(Path.of(wt).resolve("README.md"), "edited tracked file\n");
+        assertTrue(gitWorktrees.hasUncommitted(wt),
+                "a tracked modification must also count as dirty");
+    }
+
+    /**
+     * CB-576 review. {@code hasUncommitted} must tolerate a missing worktree exactly like
+     * {@code remove}: an already-gone directory holds no work to lose, and throwing here would
+     * break teardown — SessionManager.release() calls it before stopping the pane, so an
+     * exception would orphan a live pane and skip the release notification (CB-516).
+     */
+    @Test
+    void hasUncommittedOnAMissingWorktreeReturnsFalseWithoutThrowing(@TempDir Path tmp) {
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
+        String gone = tmp.resolve("wts").resolve("does-not-exist").toString();
+
+        assertFalse(gitWorktrees.hasUncommitted(gone),
+                "a missing worktree is reported clean, not an error");
+    }
+
     /** All three protected configs are covered: each one present in a worktree is neutralized and hidden. */
     @Test
     void allThreeConfigsAreNeutralizedWhenPresent(@TempDir Path tmp) throws Exception {

bridged/src/test/java/dev/ltms/bridged/session/SessionManagerTest.java

@@ -10,6 +10,7 @@ import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.WorkspaceControl;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import dev.ltms.bridged.peer.PeerUnreachableException;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
@@ -101,7 +102,7 @@ class SessionManagerTest {
 
         assertDoesNotThrow(() -> sessions.asPresence().markPresent(null),
                 "the primary's null terminal must not blow up an unrelated tool call");
-        assertDoesNotThrow(() -> sessions.onDelivered(null));
+        assertDoesNotThrow(() -> sessions.onDelivered(null, TestTurnTokens.inert(null)));
         assertDoesNotThrow(() -> sessions.onTurnComplete(null));
         assertDoesNotThrow(() -> sessions.onTurnFailed(null));
 
@@ -121,7 +122,7 @@ class SessionManagerTest {
                 "MCP presence moves SPAWNING → READY");
         assertTrue(sessions.asPresence().isPresent(terminal), "presence is also recorded");
 
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
                 "delivery moves READY → BUSY");
 
@@ -153,7 +154,7 @@ class SessionManagerTest {
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         String terminal = session.terminalId();
         sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
 
         sessions.onTurnFailed(terminal);
 
@@ -181,7 +182,7 @@ class SessionManagerTest {
             MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
             String terminal = session.terminalId();
             sessions.asPresence().markPresent(terminal);
-            sessions.onDelivered(terminal);
+            sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
 
             sessions.onTurnFailed(terminal);
 
@@ -264,7 +265,7 @@ class SessionManagerTest {
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         String terminal = session.terminalId();
         sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
 
         clock[0] = 100;
         assertEquals(0, sessions.reapIdle(10), "BUSY session past TTL is never reaped");
@@ -281,7 +282,7 @@ class SessionManagerTest {
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         String terminal = session.terminalId();
         sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         sessions.onTurnComplete(terminal);
 
         clock[0] = 21;
@@ -299,7 +300,7 @@ class SessionManagerTest {
         MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "owner2");
         sessions.asPresence().markPresent(ready.terminalId());
         sessions.asPresence().markPresent(busy.terminalId());
-        sessions.onDelivered(busy.terminalId());
+        sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
 
         clock[0] = 50;
         assertEquals(1, sessions.reapIdle(30), "only READY past TTL is reaped");
@@ -317,9 +318,9 @@ class SessionManagerTest {
         String terminal = session.terminalId();
         sessions.asPresence().markPresent(terminal);
 
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         sessions.onTurnComplete(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         sessions.onTurnComplete(terminal);
 
         MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -337,13 +338,13 @@ class SessionManagerTest {
         String terminal = session.terminalId();
         sessions.asPresence().markPresent(terminal);
 
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         sessions.onTurnComplete(terminal);
         assertEquals(MemberSession.State.DONE,
                 sessions.get(session.paneId()).orElseThrow().state(),
                 "first turn completes without release");
 
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
         sessions.onTurnComplete(terminal);
 
         assertTrue(sessions.get(session.paneId()).isEmpty(), "session released after cap reached");
@@ -359,7 +360,7 @@ class SessionManagerTest {
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         sessions.asPresence().markPresent(session.terminalId());
 
-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
         assertTrue(sessions.onTurnCompleteWithPostAction(session.terminalId()));
 
         MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -373,7 +374,7 @@ class SessionManagerTest {
         SessionManager sessions = sessionManager(herdr, () -> 0L, 1, true);
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         sessions.asPresence().markPresent(session.terminalId());
-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
 
         assertFalse(sessions.hasPostTurnAction(session.terminalId()),
                 "a session at its cap will be released, not reset for reuse");
@@ -388,7 +389,7 @@ class SessionManagerTest {
         SessionManager sessions = sessionManager(herdr, () -> 0L, 0, false);
         MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
         sessions.asPresence().markPresent(session.terminalId());
-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
 
         sessions.onTurnComplete(session.terminalId());
 
@@ -406,7 +407,7 @@ class SessionManagerTest {
         MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "ownerB");
         sessions.asPresence().markPresent(ready.terminalId());
         sessions.asPresence().markPresent(busy.terminalId());
-        sessions.onDelivered(busy.terminalId());
+        sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
 
         sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
 

bridged/src/test/java/dev/ltms/bridged/session/WorktreeSessionManagerTest.java

@@ -6,14 +6,21 @@ import dev.ltms.bridged.guard.SubscriptionGuard;
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.WorkspaceControl;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import dev.ltms.bridged.peer.MemberRole;
 import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -178,6 +185,77 @@ class WorktreeSessionManagerTest {
         assertTrue(sessions.get(paneId).isEmpty(), "released session is no longer retrievable");
     }
 
+    /**
+     * CB-576. A normal {@code COMPLETED} release whose worktree holds uncommitted work must NOT
+     * remove it — {@code --force} would destroy the worker's only copy. The bridge cannot see
+     * uncommitted files, so the worktree is preserved and the release logged at WARN naming the
+     * path, the session, and the cause an operator needs to find the work.
+     */
+    @Test
+    void releasePreservesDirtyWorktreeAndLogsWarn() {
+        FakeHerdr herdr = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
+                .withDirty(true);
+        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-576", null));
+
+        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.setContext(ctx);
+        appender.start();
+        sessionLog.addAppender(appender);
+        sessionLog.setLevel(Level.WARN);
+        try {
+            sessions.release(s.paneId());
+
+            assertTrue(herdr.called("pane.close"), "release still tears the worker pane down");
+            assertTrue(worktrees.removeCalls().isEmpty(),
+                    "a dirty worktree is never removed — it holds the only copy of the work");
+            String warn = appender.list.stream()
+                    .filter(e -> e.getLevel().equals(Level.WARN))
+                    .map(ILoggingEvent::getFormattedMessage)
+                    .filter(m -> m.contains("dirty worktree"))
+                    .findFirst()
+                    .orElse("no dirty-release WARN logged");
+            assertTrue(warn.contains(s.worktree()), "the WARN names the worktree path: " + warn);
+            assertTrue(warn.contains(s.terminalId()), "the WARN names the session: " + warn);
+            assertTrue(warn.contains("COMPLETED"), "the WARN names the release cause: " + warn);
+        } finally {
+            sessionLog.detachAppender(appender);
+        }
+    }
+
+    /**
+     * CB-576 review. A worktree that is already gone (operator cleanup, {@code git worktree prune},
+     * an earlier half-completed release) must not break teardown. {@code hasUncommitted} reports the
+     * missing path clean, so release still runs {@code notifyReleased} (the CB-516 fast-fail for a
+     * blocked {@code bridge_send} caller) and {@code launcher.stop} (so the pane is not orphaned),
+     * and falls through to the already-gone-tolerant {@code remove}.
+     */
+    @Test
+    void releaseStillStopsPaneAndNotifiesWhenWorktreeIsGone() {
+        FakeHerdr herdr = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt");
+        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-576g", null));
+        AtomicReference<String> releasedTerminal = new AtomicReference<>();
+        sessions.onRelease(releasedTerminal::set);
+
+        worktrees.markGone(s.worktree());
+        sessions.release(s.paneId());
+
+        assertEquals(s.terminalId(), releasedTerminal.get(),
+                "notifyReleased must still fire when the worktree is already gone (CB-516)");
+        assertTrue(herdr.called("pane.close"),
+                "the pane must still be stopped when the worktree is already gone");
+        assertEquals(1, worktrees.removeCalls().size(),
+                "release still calls the already-gone-tolerant remove");
+    }
+
     @Test
     void drainAllPreservesWorktreeOfIdleSession() {
         FakeHerdr herdr = new FakeHerdr();
@@ -204,7 +282,7 @@ class WorktreeSessionManagerTest {
                 new WorktreeRequest("cb-544", null));
         String terminal = s.terminalId();
         sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal); // BUSY, never completes → still BUSY when the timeout hits
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal)); // BUSY, never completes → still BUSY when the timeout hits
 
         sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
 

bridged/src/test/resources/logback-test.xml

@@ -16,6 +16,9 @@
           AuditLogTest, which attaches its own ListAppender and asserts on emitted records.
     -->
 
+    <!-- Keep the test logger behaviour aligned with the production cancellation filter. -->
+    <turboFilter class="dev.ltms.bridged.logging.McpCancelledNotificationFilter"/>
+
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
             <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
@@ -34,4 +37,4 @@
         <appender-ref ref="STDOUT"/>
     </root>
 
-</configuration>
+</configuration>

docs/M4-Fleet-Health.md

@@ -237,7 +237,7 @@ state never presents stop as the only action.
 | Release cause | Process action | Provisioned worktree |
 |---|---|---|
 | `SPAWN_ROLLBACK` before registration or delivery | Stop and clean up | Remove |
-| `COMPLETED` for `READY` or `DONE` without pending work, idle TTL, or successful context-cap completion | Stop | Remove under completed policy |
+| `COMPLETED` for `READY` or `DONE` without pending work, idle TTL, or successful context-cap completion | Stop | Remove only if clean; preserve a dirty worktree (CB-576) |
 | `NEVER_READY` | Stop | Preserve |
 | `GONE` | Best-effort stop | Preserve |
 | `TURN_FAILED` or lead abort while `BUSY` or `FAILED` | Stop | Preserve |
@@ -744,8 +744,28 @@ worktree discovery.
 
 Acceptance criteria:
 
-1. Every accepted send receives a stable `TurnToken` tied to target, exact waiter, session turn,
-   delivery baseline, and task outcome.
+1. Every accepted send receives a stable `TurnToken` tied to target, exact waiter, and delivery
+   baseline.
+
+   **Corrected during implementation (2026-08-15).** This criterion first also required the session
+   turn number and the task outcome. That is not implementable at this layer, and the implementer
+   refused it three times rather than fabricate a value — correctly. The reason is an ordering fact
+   that is invisible from any single class: `MessageService` owns acceptance and holds the waiter and
+   the async `Task`, but it learns nothing about delivery, because the delivery event goes to
+   `CompletionResolver` through `TurnListener.onDelivered`. And `CompletionResolver.onDelivered` runs
+   *before* `SessionManager.onDelivered`, so the session turn number does not exist yet at the only
+   point where the token could capture it.
+
+   Two ways out were rejected. A shared registry keyed by target reintroduces exactly the "whichever
+   send happens to be waiting" ambiguity the token exists to remove — the same weak claim
+   `Rendezvous.currentWaiter` warns about. Injecting a turn counter into `MessageService` adds a
+   required cross-layer dependency to populate a field that nothing in this slice reads, which is
+   speculative coupling across a boundary already shown to be fragile.
+
+   So the token identifies the **accepted send**, and `SessionManager` keeps verifying its own
+   delivery separately. Repair (criterion 2) does need the session turn; binding it means resolving
+   that acceptance-versus-delivery ordering first, and that work belongs to the repair unit, not
+   here. The token record carries a comment saying the field is deliberately absent.
 2. Repair requires the same `BUSY` token, two raw `IDLE` or `DONE` snapshots, no conflicting
    observation, exact open waiter, successful baseline, and new recognised assistant output.
 3. Missing, failed, late, or post-restart baseline never authorises repair.