CB-573: carry accepted turn tokens on delivery

An opt-in whole-fleet observer, separate from the 250ms delivery poller.
One AgentControl.list and one roster snapshot per tick, joined and fed to
the FleetHealth classifier, because a fault is a disagreement between the
two views at the same instant. Absent a health: block nothing is built
and no herdr call is made.

Adds bridge_list healthCoverage: off, detection-only, or full. Detection
is deliberately separate from notification, so a single-lead setup with
no webhook still gets detection and is told its coverage is partial
rather than being refused.

Two review fixes worth naming. tick() rescheduled itself as its last
statement with no try/catch, and a ScheduledExecutorService does not
re-run a task that threw — so the first agents.list failure would have
stopped health permanently and silently, which is exactly when the
control link is down. It now catches Throwable and reschedules in a
finally. And the snapshot fields this unit cannot supply are the named
constant NOT_YET_OBSERVED rather than bare false literals, because false
means no fault to this classifier.

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

@@ -9,6 +9,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -170,8 +171,8 @@ 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 tasks that have accepted delivery for a target. */
+    private final ConcurrentHashMap<String, Set<Task>> asyncTasksByTarget = 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,14 @@ public final class MessageService {
             // failed send leaves no stale waiter behind.
             CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
             try {
+                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()));
@@ -527,17 +541,18 @@ public final class MessageService {
                     if (onAccepted != null) {
                         onAccepted.run();
                     }
-                    asyncTasksByTarget.put(target, task);
+                    asyncTasksByTarget.computeIfAbsent(target, _ -> ConcurrentHashMap.newKeySet()).add(task);
                 };
-                Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted);
+                Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted, 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);
+                untrackAsyncTarget(task);
             }
         });
         pruneTerminalTickets();
@@ -600,7 +615,11 @@ 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);
+        Set<Task> targetTasks = asyncTasksByTarget.get(target);
+        Task task = targetTasks == null ? null : targetTasks.stream()
+                .filter(candidate -> !candidate.future.isDone())
+                .findFirst()
+                .orElse(null);
         if (task != null) {
             task.question = new Reply(Outcome.QUESTION, text, turnId);
             task.turnId = turnId;
@@ -623,7 +642,7 @@ 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);
+        untrackAsyncTarget(task);
         if (task.turnId != null) {
             asyncTasksByTurn.remove(task.turnId, task);
         }
@@ -642,6 +661,17 @@ public final class MessageService {
         return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
     }
 
+    /** Stop tracking a task once it no longer owns an accepted target turn. */
+    private void untrackAsyncTarget(Task task) {
+        Set<Task> targetTasks = asyncTasksByTarget.get(task.target);
+        if (targetTasks != null) {
+            targetTasks.remove(task);
+            if (targetTasks.isEmpty()) {
+                asyncTasksByTarget.remove(task.target, targetTasks);
+            }
+        }
+    }
+
     /** Release the async executor. */
     public void close() {
         asyncExecutor.shutdown();

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/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;
@@ -425,7 +426,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/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/InjectorTest.java

@@ -259,6 +259,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 +270,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.
@@ -322,6 +329,23 @@ class InjectorTest {
         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");
+        CompletableFuture<Void> queued = inj.enqueue(T, "queued");
+
+        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

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");
+        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,59 @@ 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());
+    }
+
+    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/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>