CB-568: fail queued async tickets on teardown

A worker on a wait:false delegation called bridge_ask and the lead never
saw the question. Outcome.QUESTION is deliberately non-terminal, but
taskView tested r.completed() and fell into the failure branch, so the
ticket was marked FAILED and both the question text and its turnId were
discarded. The worker blocked for 55s, gave up, and had to abandon its
task. CLAUDE.md tells leads to prefer wait:false and to answer an ask with
bridge_send{turnId, content}; those two could not both be followed.

bridge_poll now returns a non-terminal ASKING phase carrying the question
and its turnId, and the ticket stays live so the worker's real reply still
lands on it. An unanswered ask returns the ticket to PENDING, because only
the question wait ended - the delegated turn continues. The 55s/115s ask
caps are unchanged: they exist because the worker's own MCP call would time
out, so widening them would only move the failure.

Two defects found reviewing the first revision, both from replacing
supplyAsync with a manually completed future:

- an exception inside the send left the future uncompleted, so the ticket
  stayed PENDING for the life of the daemon. Now caught and completed
  exceptionally.
- correlation was keyed by target, one entry per worker, registered before
  the session lock. With two tickets outstanding on one target the second
  overwrote the first, so a late reply could resolve the wrong ticket.
  Correlation is now per turn, the target entry exists only while that send
  owns the lock, and a reply with no live waiter still goes to the durable
  inbox as before.

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

@@ -285,6 +285,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);
@@ -383,7 +389,11 @@ public final class Bridged {
         }
 
         BridgeMcp mcp = new BridgeMcp(messages, workers, sessions, identity, presence,
-                primaryRegistry, callers, metrics);
+                primaryRegistry, callers, metrics, new BridgeMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
+                profile -> {
+                    var configured = config.get().profiles().get(profile);
+                    return configured == null ? null : configured.maxLoad();
+                }, () -> config.get().profiles().keySet(), System::nanoTime));
 
         // 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.

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

@@ -0,0 +1,40 @@
+package dev.ltms.bridged.health;
+
+import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.session.MemberSession;
+
+/**
+ * Pure classifier. Collection and repair are deliberately outside this package.
+ * {@link HealthState#ERROR_ON_SCREEN} is not decided yet because it needs a bounded pane detection
+ * read and an adapter-specific fatal signature; status facts alone must not guess it.
+ */
+public final class FleetHealth {
+    private FleetHealth() { }
+
+    public static HealthDecision decide(HealthSnapshot s, HealthPrior prior, long nowNanos) {
+        if (s.controlLinkDown()) return result(HealthState.CONTROL_LINK_DOWN, false);
+        if (s.targetNotFound()) return result(HealthState.GONE, false);
+        if (s.sessionState() == MemberSession.State.SPAWNING && !s.present() && s.readinessGraceElapsed()) {
+            return result(HealthState.NEVER_READY, false);
+        }
+        if (s.orphanedDelegation()) return result(HealthState.DELEGATION_ORPHANED, false);
+        boolean disagreement = s.sessionState() == MemberSession.State.BUSY && s.acceptedDelivery()
+                && (s.liveStatus() == AgentStatus.IDLE || s.liveStatus() == AgentStatus.DONE);
+        if (disagreement && prior.busyButDone()) return result(HealthState.TURN_BOUNDARY_LOST, true);
+        if (s.stalled()) return result(HealthState.STALL_SUSPECTED, disagreement);
+        if (s.replyStranded()) return result(HealthState.REPLY_STRANDED, disagreement);
+        if (s.queuedDelivery() || s.inboxMessage()) return result(HealthState.WORK_PENDING, disagreement);
+        if (s.sessionState() == MemberSession.State.SPAWNING) return result(HealthState.STARTING, disagreement);
+        if (s.acceptedDelivery() && s.liveStatus() == AgentStatus.BLOCKED) {
+            return result(HealthState.BLOCKED_AMBIGUOUS, disagreement);
+        }
+        // An accepted delivery remains bridge work even when herdr is late, unknown, or has already
+        // reported DONE once. It cannot be IDLE until the delegation has resolved.
+        if (s.acceptedDelivery()) return result(HealthState.WORKING, disagreement);
+        return result(HealthState.IDLE, disagreement);
+    }
+
+    private static HealthDecision result(HealthState state, boolean disagreement) {
+        return new HealthDecision(state, new HealthPrior(disagreement));
+    }
+}

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

@@ -0,0 +1,4 @@
+package dev.ltms.bridged.health;
+
+/** Classification plus the private fact that the next pure decision needs. */
+public record HealthDecision(HealthState state, HealthPrior prior) { }

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

@@ -0,0 +1,6 @@
+package dev.ltms.bridged.health;
+
+/** Private cross-tick observation. It is deliberately not a reported health value. */
+public record HealthPrior(boolean busyButDone) {
+    public static final HealthPrior NONE = new HealthPrior(false);
+}

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

@@ -0,0 +1,11 @@
+package dev.ltms.bridged.health;
+
+import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.session.MemberSession;
+
+/** Read-only facts from one fleet collection tick. */
+public record HealthSnapshot(MemberSession.State sessionState, AgentStatus liveStatus,
+                             boolean acceptedDelivery, boolean queuedDelivery, boolean inboxMessage,
+                             boolean present, boolean targetNotFound, boolean controlLinkDown,
+                             boolean readinessGraceElapsed, boolean orphanedDelegation,
+                             boolean replyStranded, boolean stalled) { }

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

@@ -0,0 +1,8 @@
+package dev.ltms.bridged.health;
+
+/** Health classifications reported for a member. */
+public enum HealthState {
+    STARTING, IDLE, WORKING, WORK_PENDING, BLOCKED_AMBIGUOUS,
+    NEVER_READY, GONE, TURN_BOUNDARY_LOST, ERROR_ON_SCREEN, STALL_SUSPECTED,
+    MUTE, REPLY_STRANDED, DELEGATION_ORPHANED, CONTROL_LINK_DOWN
+}

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

@@ -0,0 +1,25 @@
+package dev.ltms.bridged.health;
+
+import dev.ltms.bridged.msg.Rendezvous;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Counts turns that ended via the completion fallback instead of {@code bridge_reply}.
+ * MUTE is an observation by target and profile, not a classifier state and never suppresses faults.
+ */
+public final class MuteCounter {
+    private final Map<String, Integer> byTarget = new ConcurrentHashMap<>();
+    private final Map<String, Integer> byProfile = new ConcurrentHashMap<>();
+
+    /** Record only fallback completion; a structured reply does not make a member mute. */
+    public void observe(String target, String profile, Rendezvous.Kind kind) {
+        if (kind != Rendezvous.Kind.COMPLETION) return;
+        byTarget.merge(target, 1, Integer::sum);
+        byProfile.merge(profile, 1, Integer::sum);
+    }
+
+    public int forTarget(String target) { return byTarget.getOrDefault(target, 0); }
+    public int forProfile(String profile) { return byProfile.getOrDefault(profile, 0); }
+}

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

@@ -0,0 +1,26 @@
+package dev.ltms.bridged.health;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Fixed pane-probe limits. Pane content is never retained here. */
+public final class PaneBudget {
+    public static final long COOLDOWN_NANOS = 60_000_000_000L;
+    public static final int MAX_PER_TICK = 2;
+    private final Map<String, Long> lastProbe = new HashMap<>();
+    private int cursor;
+
+    public List<String> choose(List<String> candidates, long nowNanos, long configuredCooldownNanos) {
+        long cooldown = Math.max(COOLDOWN_NANOS, configuredCooldownNanos);
+        List<String> out = new ArrayList<>();
+        for (int n = 0; n < candidates.size() && out.size() < MAX_PER_TICK; n++) {
+            String target = candidates.get((cursor + n) % candidates.size());
+            Long last = lastProbe.get(target);
+            if (last == null || nowNanos - last >= cooldown) { out.add(target); lastProbe.put(target, nowNanos); }
+        }
+        if (!candidates.isEmpty()) cursor = (cursor + 1) % candidates.size();
+        return List.copyOf(out);
+    }
+}

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

@@ -137,7 +137,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 +198,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 +212,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

@@ -410,8 +410,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

@@ -41,6 +41,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

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

@@ -33,7 +33,10 @@ import jakarta.servlet.http.HttpServlet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 /**
@@ -73,15 +76,14 @@ public final class BridgeMcp {
     private final McpSyncServer server;
     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;
 
-    /**
-     * Legacy constructor — no authorization. Retained so existing tests exercise tool behaviour
-     * without an auth fixture.
-     */
-    public BridgeMcp(MessageService messages, PeerLauncher workers,
-                     SessionManager sessions, ConnectionIdentity identity, MemberPresence presence,
-                     PrimaryRegistry primaryRegistry) {
-        this(messages, workers, sessions, identity, presence, primaryRegistry, null, null);
+    /** 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,
+                                 Supplier<Set<String>> configuredProfiles, LongSupplier clock) {
+        /** Inert test-only source. It omits capacity rather than inventing zero live counts. */
+        public static CapacitySource none() { return new CapacitySource(_ -> 0, _ -> null, Set::of, System::nanoTime); }
+        boolean available() { return !configuredProfiles.get().isEmpty(); }
     }
 
     /**
@@ -91,9 +93,10 @@ public final class BridgeMcp {
      *                filter, so the REST guard does not cover it.
      * @param metrics registry for auth-failure counting; may be {@code null}
      */
-    public BridgeMcp(MessageService messages, PeerLauncher workers,
-                     SessionManager sessions, ConnectionIdentity identity, MemberPresence presence,
-                     PrimaryRegistry primaryRegistry, CallerResolver callers, Metrics metrics) {
+    public BridgeMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
+                     ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
+                     CallerResolver callers, Metrics metrics, CapacitySource capacity) {
+        this.capacity = capacity;
         McpJsonMapper json = new JacksonMcpJsonMapperSupplier().get();
         this.transport = HttpServletStreamableServerTransportProvider.builder()
                 .jsonMapper(json)
@@ -151,8 +154,8 @@ public final class BridgeMcp {
                     Runnable onAccepted = () -> primaryRegistry.recordDelegation(target, caller);
                     // wait defaults to true (block for the reply); wait:false is fire-and-poll.
                     return Boolean.FALSE.equals(a.get("wait"))
-                            ? sendAsync(messages, target, content, onAccepted)
-                            : send(messages, target, content, timeoutMs(a), onAccepted);
+                            ? sendAsync(messages, target, content, onAccepted, workers.profiles())
+                            : send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles());
                 })
                 // bridge_reply's identity is the CONNECTION, never an argument — so the authz check
                 // is "is this caller a worker at all", and it can only ever reply as itself.
@@ -207,7 +210,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,
+                    return listFleet(workers, sessions, messages, capacity,
                             callers == null ? Map.of() : callers.leads(),
                             callerTerminal(exchange));
                 })
@@ -379,21 +382,22 @@ public final class BridgeMcp {
 
     // --- tool logic (thin adapters over the services; unit-testable) ---------------------------
 
-    /** {@code bridge_send}: delegate {@code content} to a worker session and block for its reply. */
-    static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content, Long timeoutMs) {
-        return send(messages, sessionId, content, timeoutMs, null);
-    }
-
     /**
-     * As {@link #send(MessageService, String, String, Long)}, wiring an accepted-delivery hook
+     * {@code bridge_send}: delegate {@code content} to a worker session and block for its reply.
+     * The configured profiles are required so a profile name can never bypass target validation.
+     *
      * (CB-548): {@code onAccepted} records delegator ownership the instant the send is accepted, so
      * a BUSY interloper never claims a turn it did not win. {@code null} disables recording.
      */
     static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content,
-                                         Long timeoutMs, Runnable onAccepted) {
+                                         Long timeoutMs, Runnable onAccepted, Set<String> profiles) {
         if (isBlank(sessionId) || isBlank(content)) {
             return error("sessionId and content are required");
         }
+        McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
+        if (targetError != null) {
+            return targetError;
+        }
         long timeout = clamp(timeoutMs == null ? DEFAULT_TIMEOUT_MS : timeoutMs);
         try {
             return formatReply(messages.send(sessionId, content, timeout, onAccepted), timeout);
@@ -463,24 +467,33 @@ public final class BridgeMcp {
     /**
      * {@code bridge_send} with {@code wait:false}: delegate {@code content} and return a ticket
      * immediately (fire-and-poll), so a long task isn't cut off by the caller's MCP call timeout.
-     */
-    static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content) {
-        return sendAsync(messages, sessionId, content, null);
-    }
-
-    /**
-     * As {@link #sendAsync(MessageService, String, String)}, wiring the accepted-delivery hook
+     * The configured profiles are required so a profile name can never bypass target validation.
+     *
+     * This wires the accepted-delivery hook
      * (CB-548) so an async flooding send records delegator ownership exactly once it is accepted.
      */
     static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content,
-                                              Runnable onAccepted) {
+                                              Runnable onAccepted, Set<String> profiles) {
         if (isBlank(sessionId) || isBlank(content)) {
             return error("sessionId and content are required");
         }
+        McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
+        if (targetError != null) {
+            return targetError;
+        }
         String ticket = messages.sendAsync(sessionId, content, onAccepted);
         return text("accepted — task delegated. Poll bridge_poll with ticket=" + ticket);
     }
 
+    /** A configured profile is never a send target; other unknown values may be herdr-owned panes. */
+    private static McpSchema.CallToolResult profileTargetError(String sessionId, Set<String> profiles) {
+        if (profiles.contains(sessionId)) {
+            return error("unknown send target \"" + sessionId + "\": it is a configured profile name, not a "
+                    + "session id. Call bridge_list to find a member or lead sessionId.");
+        }
+        return null;
+    }
+
     /** {@code bridge_poll}: check an async delegation by ticket, or drain a worker's inbox by target. */
     static McpSchema.CallToolResult poll(MessageService messages, String ticket, String target) {
         if (!isBlank(target)) {
@@ -502,6 +515,9 @@ public final class BridgeMcp {
                     ? "[done — worker finished without a structured bridge_reply; transcript tail follows]\n" + v.reply()
                     : v.reply());
             case PENDING -> text("[pending — " + v.detail() + "]");
+            case ASKING -> text("[question — worker is waiting for your answer]\n" + v.reply()
+                    + "\n\nAnswer it by calling bridge_send again with turnId=\"" + v.turnId()
+                    + "\" and content set to your answer; the worker resumes the same turn.");
             case FAILED -> text("[failed — " + v.detail() + "]");
         };
     }
@@ -708,6 +724,12 @@ 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);
+    }
+
+    static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
+                                              CapacitySource capacity,
+                                              Map<String, String> leads, String selfTerm) {
         try {
             Map<String, Agent> live = workers.list().stream()
                     .map(Agent.class::cast)
@@ -717,15 +739,56 @@ public final class BridgeMcp {
                     .sorted(Map.Entry.comparingByValue())
                     .map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm))
                     .toList();
-            List<Map<String, Object>> out = sessions.roster().stream()
-                    .map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
+            List<MemberSession> roster = sessions.roster();
+            List<Map<String, Object>> out = roster.stream()
+                    .map(s -> memberCapacityView(s, live.get(s.terminalId()), messages, capacity.clock().getAsLong()))
                     .toList();
-            return text(json(Map.of("leads", leadRows, "members", out)));
+            Set<String> profiles = new java.util.TreeSet<>(capacity.configuredProfiles().get());
+            roster.stream().map(MemberSession::profile).forEach(profiles::add);
+            Map<String, Object> result = new LinkedHashMap<>();
+            result.put("leads", leadRows); result.put("members", out);
+            if (capacity.available()) result.put("capacity", profiles.stream()
+                    .map(profile -> capacityView(profile, capacity.liveCount(), capacity.maxLoad(), roster, messages,
+                            capacity.clock().getAsLong())).toList());
+            return text(json(result));
         } catch (HerdrException e) {
             return error("herdr error listing the fleet: " + e.getMessage());
         }
     }
 
+    /**
+     * Capacity is advisory only. {@code reclaimable} says there is no bridge work, not that bridged
+     * may stop the member: the bridge has capacity facts but no work list, and choosing work needs
+     * authority it does not have. {@code idleForSeconds} is derived from monotonic nanoTime and has
+     * no wall-clock meaning across a daemon restart.
+     */
+    private static Map<String, Object> memberCapacityView(MemberSession session, Agent live,
+                                                           MessageService messages, long nowNanos) {
+        Map<String, Object> row = SessionManager.rosterView(session, live);
+        boolean open = messages != null && messages.hasAcceptedDelivery(session.terminalId());
+        boolean inbox = messages != null && messages.hasInboxMessage(session.terminalId());
+        boolean reclaimable = (session.state() == MemberSession.State.READY || session.state() == MemberSession.State.DONE)
+                && !open && !inbox;
+        row.put("reclaimable", reclaimable);
+        row.put("idleForSeconds", reclaimable ? Math.max(0, (nowNanos - session.lastActivityAtNanos()) / 1_000_000_000L) : null);
+        return row;
+    }
+
+    private static Map<String, Object> capacityView(String profile, Function<String, Integer> liveCount,
+                                                     Function<String, Integer> maxLoad, List<MemberSession> roster,
+                                                     MessageService messages, long nowNanos) {
+        Integer cap = maxLoad.apply(profile);
+        int live = liveCount.apply(profile);
+        int reclaimable = (int) roster.stream().filter(s -> profile.equals(s.profile()))
+                .filter(s -> (s.state() == MemberSession.State.READY || s.state() == MemberSession.State.DONE))
+                .filter(s -> messages == null || (!messages.hasAcceptedDelivery(s.terminalId()) && !messages.hasInboxMessage(s.terminalId())))
+                .count();
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("profile", profile); row.put("maxLoad", cap); row.put("live", live);
+        row.put("free", cap == null ? null : Math.max(0, cap - live)); row.put("reclaimable", reclaimable);
+        return row;
+    }
+
     /**
      * One lead's row: its address, its name, and whether it can be reached right now.
      *

bridged/src/main/java/dev/ltms/bridged/member/ClaudeCodeLauncher.java

@@ -181,8 +181,8 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
         // id via -r and passes no --session-id (the two conflict). Both are injected before the
         // model flag so --model keeps outranking the operator's own argv.
         // mutableArgv: argvWithBridge may hand back the profile's own (immutable) List.of when it
-        // has no MCP — session flags must be added into a list we own.
-        List<String> argv = mutableArgv(argvWithBridge(cfg));
+        // has neither MCP nor a charter — session flags must be added into a list we own.
+        List<String> argv = mutableArgv(argvWithBridge(cfg, spec.charter()));
         String agentSessionId = applySessionIdentity(argv, spec.sessionName(), spec.resumeSessionId());
         return new Launch(workerEnv, argvWithModel(argv, cfg), agentSessionId);
     }
@@ -217,22 +217,26 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
     }
 
     /**
-     * The launch argv, plus — when {@code worker.mcpUrl} is set — inline {@code --mcp-config} for
-     * the bridge server and {@code --append-system-prompt} for the {@link #REPLY_CHARTER}. Neither
-     * touches the profile's config; both are pure command-line flags. This inline-flag mount is
-     * Claude Code specific — other adapters mount MCP and instructions their own way.
+     * The launch argv, plus an inline {@code --mcp-config} when {@code worker.mcpUrl} is set and
+     * {@code --append-system-prompt} when the base composed a charter. Neither touches the profile's
+     * config; both are pure command-line flags. This inline-flag mount is Claude Code specific —
+     * other adapters mount MCP and instructions their own way.
      */
-    private List<String> argvWithBridge(BridgedConfig.Profile cfg) {
-        if (!cfg.hasMcp()) {
+    private List<String> argvWithBridge(BridgedConfig.Profile cfg, String charter) {
+        if (!cfg.hasMcp() && charter == null) {
             return cfg.argv();
         }
-        String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
-                + cfg.mcpUrl() + "\"}}}";
         List<String> argv = mutableArgv(cfg.argv());
-        argv.add("--mcp-config");
-        argv.add(mcpJson);
-        argv.add("--append-system-prompt");
-        argv.add(REPLY_CHARTER);
+        if (cfg.hasMcp()) {
+            String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
+                    + cfg.mcpUrl() + "\"}}}";
+            argv.add("--mcp-config");
+            argv.add(mcpJson);
+        }
+        if (charter != null) {
+            argv.add("--append-system-prompt");
+            argv.add(charter);
+        }
         return argv;
     }
 

bridged/src/main/java/dev/ltms/bridged/member/OpenCodeLauncher.java

@@ -163,10 +163,10 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
      * {@inheritDoc}
      *
      * <p>Builds the opencode launch: no {@code ANTHROPIC_*} and no guard (opencode reads its own
-     * provider credentials); when the profile mounts the bridge MCP or has a member charter, generate an ephemeral
-     * {@code opencode.json} (remote MCP server + member-charter instructions) and point the worker at
-     * it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge grant; and select the model
-     * with {@code -m}.
+     * provider credentials); when the profile mounts the bridge MCP or has a member charter,
+     * generate an ephemeral {@code opencode.json} (remote MCP server + member-charter instructions)
+     * and point the worker at it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge
+     * grant; and select the model with {@code -m}.
      */
     @Override
     protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {

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;
@@ -127,6 +128,8 @@ public final class MessageService {
     public enum Phase {
         /** Delegated and in flight — queued for the worker or being worked. */
         PENDING,
+        /** The worker is paused in {@code bridge_ask}; {@link TaskView#reply} and {@link TaskView#turnId} identify it. */
+        ASKING,
         /** The worker's turn finished; {@link TaskView#reply} holds the answer. */
         DONE,
         /** The delegation could not complete (timed out, worker gone, or busy). */
@@ -136,16 +139,28 @@ public final class MessageService {
     /**
      * A poll snapshot of an async delegation.
      *
-     * @param reply       the answer when {@link #phase} is {@link Phase#DONE}, else {@code null}
+     * @param reply       the answer when {@link #phase} is {@link Phase#DONE}, or the question when
+     *                    {@link #phase} is {@link Phase#ASKING}; otherwise {@code null}
      * @param replySource {@code "reply"} (structured {@code bridge_reply}) or {@code "transcript"}
      *                    (completion scrape) when {@link Phase#DONE}, else {@code null}
-     * @param detail      a human note (live worker status while pending, or the failure reason)
+     * @param detail      a human note (live worker status while pending, ask state, or failure reason)
+     * @param turnId      correlation id for an {@link Phase#ASKING} ticket, else {@code null}
      */
-    public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail) {
+    public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail,
+                           String turnId) {
     }
 
     /** An in-flight or finished async delegation, keyed by its ticket. */
-    private record Task(String target, CompletableFuture<Reply> future, long createdNanos) {
+    private static final class Task {
+        private final String target;
+        private final CompletableFuture<Reply> future = new CompletableFuture<>();
+        private final long createdNanos = System.nanoTime();
+        private volatile Reply question;
+        private volatile String turnId;
+
+        private Task(String target) {
+            this.target = target;
+        }
     }
 
     private final AgentControl agents;
@@ -156,6 +171,10 @@ 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<>();
+    /** 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();
     private final ExecutorService asyncExecutor = Executors.newThreadPerTaskExecutor(
             Thread.ofVirtual().name("bridge-async-", 0).factory());
@@ -202,6 +221,16 @@ public final class MessageService {
         return agents.status(target);
     }
 
+    /** Read-only delegation fact for fleet views. */
+    public boolean hasAcceptedDelivery(String target) {
+        return rendezvous.isWaiting(target);
+    }
+
+    /** Read-only inbox fact for fleet views. */
+    public boolean hasInboxMessage(String target) {
+        return !inbox.peek(target).isEmpty();
+    }
+
     /**
      * Route a worker's explicit {@code bridge_reply}: resolve an open send, or queue it in the
      * inbox if no send is currently open. Unlike the bare {@link Rendezvous#resolve}, a no-waiter
@@ -272,14 +301,14 @@ 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 = rendezvous.resolveFailure(waiter, reason);
+        boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason);
+        boolean asyncFailed = tasks.values().stream()
+                .filter(task -> target.equals(task.target) && task.question == null)
+                .anyMatch(task -> task.future.complete(new Reply(Outcome.WORKER_FAILED, reason)));
         if (failed) {
             log.warn("abandoning the blocked send to {}: {}", target, reason);
         }
-        return failed;
+        return failed || asyncFailed;
     }
 
     /**
@@ -326,6 +355,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());
 
@@ -333,6 +367,12 @@ 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
+            }
             // Open the waiter BEFORE queueing delivery (CB-548). A fast reply — the worker already
             // injectable the instant we enqueue — otherwise arrives before the waiter is registered
             // and orphans into the inbox while this send blocks to the timeout (the enqueue-before-
@@ -393,12 +433,14 @@ public final class MessageService {
                 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);
             return new AskResult(AskOutcome.ANSWERED, answer);
         } catch (TimeoutException e) {
             log.debug("bridge_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
+            clearAsyncQuestion(ticket.turnId(), true);
             return new AskResult(AskOutcome.TIMED_OUT, null);
         } catch (ExecutionException e) {
             Throwable cause = e.getCause();
@@ -441,9 +483,12 @@ public final class MessageService {
                 rendezvous.close(workerSession, reply);
                 return new Reply(Outcome.STALE_TURN, null); // lapsed between the lookup and the unblock
             }
+            clearAsyncQuestion(turnId, false);
             try {
                 Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
-                return new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
+                Reply result = new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
+                finishAsyncTask(turnId, result);
+                return result;
             } catch (TimeoutException e) {
                 // The worker resumed but hasn't replied yet — no completion fallback arms an answered
                 // turn (it never re-entered the injector), so a silent worker rides out the window.
@@ -483,9 +528,28 @@ public final class MessageService {
      */
     public String sendAsync(String target, String content, Runnable onAccepted) {
         String ticket = "task-" + ticketSeq.incrementAndGet();
-        CompletableFuture<Reply> future = CompletableFuture.supplyAsync(
-                () -> send(target, content, ASYNC_TIMEOUT_MS, onAccepted), asyncExecutor);
-        tasks.put(ticket, new Task(target, future, System.nanoTime()));
+        Task task = new Task(target);
+        tasks.put(ticket, task);
+        asyncExecutor.submit(() -> {
+            try {
+                Runnable trackingAccepted = () -> {
+                    if (onAccepted != null) {
+                        onAccepted.run();
+                    }
+                    asyncTasksByTarget.computeIfAbsent(target, _ -> ConcurrentHashMap.newKeySet()).add(task);
+                };
+                Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted, task);
+                if (result.outcome() == Outcome.QUESTION) {
+                    // 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);
+                untrackAsyncTarget(task);
+            }
+        });
         pruneTerminalTickets();
         log.debug("async send {} -> {}", ticket, target);
         return ticket;
@@ -501,27 +565,32 @@ public final class MessageService {
         if (task == null) {
             return null;
         }
-        CompletableFuture<Reply> f = task.future();
+        CompletableFuture<Reply> f = task.future;
         if (!f.isDone()) {
-            return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target()));
+            Reply question = task.question;
+            if (question != null) {
+                return new TaskView(ticket, Phase.ASKING, question.text(), null,
+                        "worker is waiting for your answer", question.turnId());
+            }
+            return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target), null);
         }
         Reply r;
         try {
             r = f.getNow(null);
         } catch (CompletionException | java.util.concurrent.CancellationException e) {
             Throwable cause = (e instanceof CompletionException ce && ce.getCause() != null) ? ce.getCause() : e;
-            return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage());
+            return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage(), null);
         }
         if (r.completed()) {
             String source = r.outcome() == Outcome.REPLIED ? "reply" : "transcript";
-            return new TaskView(ticket, Phase.DONE, r.text(), source, null);
+            return new TaskView(ticket, Phase.DONE, r.text(), source, null, null);
         }
         // A wedged worker (CB-109) carries the error context as its reason; the timeout/busy
         // outcomes carry none, so fall back to the outcome name.
         String detail = r.outcome() == Outcome.WORKER_FAILED && r.text() != null
                 ? r.text()
                 : "no reply — " + r.outcome().name().toLowerCase();
-        return new TaskView(ticket, Phase.FAILED, null, null, detail);
+        return new TaskView(ticket, Phase.FAILED, null, null, detail, null);
     }
 
     /** Best-effort live worker status for a pending poll; never throws (a lookup error is just noise). */
@@ -536,7 +605,66 @@ public final class MessageService {
     /** Drop finished tickets older than the TTL so the registry cannot grow without bound. */
     private void pruneTerminalTickets() {
         long cutoff = System.nanoTime() - TICKET_TTL_NANOS;
-        tasks.values().removeIf(t -> t.future().isDone() && t.createdNanos() < cutoff);
+        tasks.values().removeIf(t -> t.future.isDone() && t.createdNanos < cutoff);
+    }
+
+    /** 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) {
+        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;
+            asyncTasksByTurn.put(turnId, task);
+        }
+    }
+
+    /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
+    private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
+        Task task = asyncTasksByTurn.get(turnId);
+        if (task != null && turnId.equals(task.turnId)) {
+            task.question = null;
+            if (forgetTurn) {
+                asyncTasksByTurn.remove(turnId, task);
+                task.turnId = null;
+            }
+        }
+    }
+
+    /** Complete and detach an async ticket after its worker's actual terminal reply. */
+    private void finishAsyncTask(Task task, Reply result) {
+        task.future.complete(result);
+        untrackAsyncTarget(task);
+        if (task.turnId != null) {
+            asyncTasksByTurn.remove(task.turnId, task);
+        }
+    }
+
+    /** Complete the async ticket correlated to a specific answered turn. */
+    private void finishAsyncTask(String turnId, Reply result) {
+        Task task = asyncTasksByTurn.get(turnId);
+        if (task != null) {
+            finishAsyncTask(task, result);
+        }
+    }
+
+    /** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
+    private boolean hasAsyncQuestion(String target) {
+        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. */

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

@@ -0,0 +1,56 @@
+package dev.ltms.bridged.health;
+
+import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.session.MemberSession;
+import dev.ltms.bridged.msg.Rendezvous;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class FleetHealthTest {
+    @Test void muteCountsOnlyCompletionFallbacks() {
+        MuteCounter mute = new MuteCounter();
+        mute.observe("target", "terra", Rendezvous.Kind.REPLY);
+        mute.observe("target", "terra", Rendezvous.Kind.COMPLETION);
+        assertEquals(1, mute.forTarget("target"));
+        assertEquals(1, mute.forProfile("terra"));
+    }
+    @Test void turnBoundaryNeedsTwoSnapshots() {
+        HealthSnapshot s = snapshot(MemberSession.State.BUSY, AgentStatus.DONE, true);
+        HealthDecision first = FleetHealth.decide(s, HealthPrior.NONE, 1);
+        assertEquals(HealthState.WORKING, first.state());
+        assertEquals(new HealthPrior(true), first.prior());
+        assertEquals(HealthState.TURN_BOUNDARY_LOST, FleetHealth.decide(s, first.prior(), 2).state());
+    }
+
+    @Test void unknownLiveStatusWithAcceptedDeliveryIsNotIdle() {
+        assertEquals(HealthState.WORKING, FleetHealth.decide(
+                snapshot(MemberSession.State.BUSY, AgentStatus.UNKNOWN, true), HealthPrior.NONE, 1).state());
+    }
+
+    @Test void acceptedDeliveryNeverReportsIdle() {
+        for (MemberSession.State session : MemberSession.State.values()) {
+            for (AgentStatus live : AgentStatus.values()) {
+                HealthSnapshot s = snapshot(session, live, true);
+                assertEquals(false, FleetHealth.decide(s, HealthPrior.NONE, 1).state() == HealthState.IDLE,
+                        () -> "accepted delivery returned IDLE for " + session + "/" + live);
+            }
+        }
+    }
+
+    @Test void blockedDoesNotGuessPromptKind() {
+        assertEquals(HealthState.BLOCKED_AMBIGUOUS, FleetHealth.decide(
+                snapshot(MemberSession.State.BUSY, AgentStatus.BLOCKED, true), HealthPrior.NONE, 1).state());
+    }
+
+    @Test void controlLinkOutranksMemberFault() {
+        HealthSnapshot s = new HealthSnapshot(MemberSession.State.BUSY, AgentStatus.DONE, true, false,
+                false, true, true, true, false, true, true, true);
+        assertEquals(HealthState.CONTROL_LINK_DOWN, FleetHealth.decide(s, HealthPrior.NONE, 1).state());
+    }
+
+    private static HealthSnapshot snapshot(MemberSession.State state, AgentStatus live, boolean accepted) {
+        return new HealthSnapshot(state, live, accepted, false, false, true, false, false,
+                false, false, false, false);
+    }
+}

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

@@ -0,0 +1,14 @@
+package dev.ltms.bridged.health;
+
+import org.junit.jupiter.api.Test;
+import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class PaneBudgetTest {
+    @Test void fixedLimitsIgnoreWeakerConfig() {
+        PaneBudget budget = new PaneBudget();
+        List<String> targets = List.of("a", "b", "c");
+        assertEquals(List.of("a", "b"), budget.choose(targets, 0, 0));
+        assertEquals(List.of("c"), budget.choose(targets, 1, 0));
+    }
+}

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/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);
+                metrics, BridgeMcp.CapacitySource.none());
         return mcp;
     }
 

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

@@ -53,6 +53,18 @@ class BridgeMcpTest {
         return ((McpSchema.TextContent) r.content().getFirst()).text();
     }
 
+    private void assertSendRoundTrips(String target, Set<String> profiles) throws Exception {
+        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.send(messages, target, "hi", 4000L, null, profiles));
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting(target) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting(target), "send should be accepted for " + target);
+        BridgeMcp.reply(messages, target, "received");
+        assertEquals("received", textOf(send.get(6, TimeUnit.SECONDS)));
+    }
+
     private static ClaudeCodeLauncher workerService(FakeHerdr h, String baseUrl, Set<String> allow) {
         BridgedConfig.Profile cfg = new BridgedConfig.Profile(
                 "ltms-local", baseUrl, "coder", null, "BRIDGED_WORKER_TOKEN", null,
@@ -69,7 +81,7 @@ class BridgeMcpTest {
     void sendThenReplyRoundTrips() throws Exception {
         // bridge_send blocks; bridge_reply resolves it with the worker's structured answer.
         CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> BridgeMcp.send(messages, "term_a", "review this", 4000L));
+                () -> BridgeMcp.send(messages, "term_a", "review this", 4000L, null, Set.of()));
 
         // Wait until the send has opened its waiter so the reply resolves it (CB-307: reply now
         // queues in the inbox if no waiter is open, which would break the round-trip).
@@ -91,7 +103,7 @@ class BridgeMcpTest {
     @Test
     void asyncSendReturnsATicketThenPollReportsTheReply() throws Exception {
         // wait:false parity — a ticket is issued, resolved by a reply, and surfaced by bridge_poll.
-        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it");
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
         assertNotEquals(Boolean.TRUE, accepted.isError());
         String out = textOf(accepted);
         assertTrue(out.contains("ticket="), out);
@@ -120,6 +132,131 @@ class BridgeMcpTest {
         assertEquals("async LGTM", textOf(polled));
     }
 
+    @Test
+    void asyncSendSurfacesAnAskThenKeepsTheTicketForTheFinalReply() throws Exception {
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
+        String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
+        McpSchema.CallToolResult question = BridgeMcp.poll(messages, ticket, null);
+        deadline = System.currentTimeMillis() + 3000;
+        while (!textOf(question).contains("[question") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            question = BridgeMcp.poll(messages, ticket, null);
+        }
+        assertTrue(textOf(question).contains("which config?"), textOf(question));
+        String questionText = textOf(question);
+        String afterTurnId = questionText.substring(questionText.indexOf("turnId=\"") + "turnId=\"".length());
+        String turnId = afterTurnId.substring(0, afterTurnId.indexOf('"'));
+
+        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.answer(messages, turnId, "config.yaml", 5000L));
+        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
+
+        deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+        BridgeMcp.reply(messages, "term_a", "done");
+        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
+
+        McpSchema.CallToolResult done = BridgeMcp.poll(messages, ticket, null);
+        deadline = System.currentTimeMillis() + 3000;
+        while (!"done".equals(textOf(done)) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            done = BridgeMcp.poll(messages, ticket, null);
+        }
+        assertEquals("done", textOf(done));
+    }
+
+    @Test
+    void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
+        String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        McpSchema.CallToolResult ask = BridgeMcp.ask(messages, "term_a", "still there?", 50L);
+        assertTrue(textOf(ask).contains("no answer"), textOf(ask));
+        assertTrue(textOf(BridgeMcp.poll(messages, ticket, null)).startsWith("[pending"));
+
+        BridgeMcp.reply(messages, "term_a", "finished after timeout");
+        assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
+    }
+
+    @Test
+    void asyncSendFailureDoesNotLeaveItsTicketPending() throws Exception {
+        String ticket = messages.sendAsync("term_a", "do it", () -> {
+            throw new IllegalStateException("accept failed");
+        });
+
+        long deadline = System.currentTimeMillis() + 3000;
+        MessageService.TaskView view = messages.poll(ticket);
+        while (view.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            view = messages.poll(ticket);
+        }
+        assertEquals(MessageService.Phase.FAILED, view.phase());
+        assertEquals("accept failed", view.detail());
+    }
+
+    @Test
+    void anotherAsyncTicketCannotCaptureAReplyWhileTheFirstTicketIsAsking() throws Exception {
+        McpSchema.CallToolResult firstAccepted = BridgeMcp.sendAsync(messages, "term_a", "first", null, Set.of());
+        String firstTicket = textOf(firstAccepted).substring(textOf(firstAccepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
+        MessageService.TaskView first = messages.poll(firstTicket);
+        deadline = System.currentTimeMillis() + 3000;
+        while (first.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            first = messages.poll(firstTicket);
+        }
+        assertEquals(MessageService.Phase.ASKING, first.phase());
+        String firstTurnId = first.turnId();
+
+        McpSchema.CallToolResult secondAccepted = BridgeMcp.sendAsync(messages, "term_a", "second", null, Set.of());
+        String secondTicket = textOf(secondAccepted).substring(textOf(secondAccepted).indexOf("ticket=") + "ticket=".length()).trim();
+        MessageService.TaskView second = messages.poll(secondTicket);
+        deadline = System.currentTimeMillis() + 3000;
+        while (second.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            second = messages.poll(secondTicket);
+        }
+        assertEquals(MessageService.Phase.FAILED, second.phase());
+
+        BridgeMcp.reply(messages, "term_a", "late reply");
+        assertEquals("late reply", messages.drainReplies("term_a").getFirst().content());
+
+        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.answer(messages, firstTurnId, "config.yaml", 5000L));
+        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        BridgeMcp.reply(messages, "term_a", "done");
+        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
+    }
+
     @Test
     void pollUnknownTicketIsAnError() {
         McpSchema.CallToolResult res = BridgeMcp.poll(messages, "task-999", null);
@@ -129,15 +266,40 @@ class BridgeMcpTest {
 
     @Test
     void sendTimesOutWithAWorkingNote() {
-        McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L);
+        McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L, null, Set.of());
         assertNotEquals(Boolean.TRUE, res.isError(), "a timeout is informational, not a tool error");
         assertTrue(textOf(res).contains("no reply"), "got: " + textOf(res));
     }
 
     @Test
     void sendRejectsMissingArgs() {
-        assertTrue(BridgeMcp.send(messages, null, "hi", null).isError());
-        assertTrue(BridgeMcp.send(messages, "term_a", "  ", null).isError());
+        assertTrue(BridgeMcp.send(messages, null, "hi", null, null, Set.of()).isError());
+        assertTrue(BridgeMcp.send(messages, "term_a", "  ", null, null, Set.of()).isError());
+    }
+
+    @Test
+    void sendRejectsAConfiguredProfileNameBeforeAcceptingIt() {
+        McpSchema.CallToolResult blocking = BridgeMcp.send(messages, "sol", "hi", 100L, null, Set.of("sol"));
+        McpSchema.CallToolResult async = BridgeMcp.sendAsync(messages, "sol", "hi", null, Set.of("sol"));
+
+        assertTrue(blocking.isError());
+        assertTrue(async.isError());
+        assertTrue(textOf(blocking).contains("sol"));
+        assertTrue(textOf(blocking).contains("configured profile name"));
+        assertTrue(textOf(blocking).contains("bridge_list"));
+        assertFalse(textOf(async).contains("ticket="));
+    }
+
+    @Test
+    void sendAllowsPeerLeadMemberAndUnclassifiedTargets() throws Exception {
+        Set<String> profiles = Set.of("sol");
+
+        assertSendRoundTrips("term_peer_lead", profiles);
+        assertSendRoundTrips("term_live_member", profiles);
+
+        // A herdr-owned pane outside the bridge roster cannot be classified at accept time.
+        McpSchema.CallToolResult result = BridgeMcp.send(messages, "external-pane", "hi", 10L, null, profiles);
+        assertFalse(result.isError(), "an unclassified target must not be rejected at acceptance time");
     }
 
     @Test
@@ -173,7 +335,7 @@ class BridgeMcpTest {
     void askThenAnswerRoundTrips() throws Exception {
         // The primary delegates and blocks; wait until its waiter is open before the worker asks.
         CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> BridgeMcp.send(messages, "term_a", "do X", 5000L));
+                () -> BridgeMcp.send(messages, "term_a", "do X", 5000L, null, Set.of()));
         long deadline = System.currentTimeMillis() + 3000;
         while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
             //noinspection BusyWait
@@ -305,6 +467,42 @@ class BridgeMcpTest {
         assertTrue(out.contains("\"liveStatus\":\"unknown\""), out);
     }
 
+    @Test
+    void capacityUsesThePlacementLiveCount() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        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(), "");
+        String out = textOf(res);
+        assertTrue(out.contains("\"maxLoad\":2"), out);
+        assertTrue(out.contains("\"live\":2"), out);
+        assertTrue(out.contains("\"free\":0"), out);
+    }
+
+    @Test
+    void capacityIncludesConfiguredProfileWithoutMembers() {
+        FakeHerdr h = new FakeHerdr();
+        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(), ""));
+        assertTrue(out.contains("\"profile\":\"terra\""), out);
+        assertTrue(out.contains("\"live\":0"), out);
+        assertTrue(out.contains("\"free\":2"), out);
+        assertTrue(out.contains("\"reclaimable\":0"), out);
+    }
+
+    @Test
+    void inertCapacitySourceOmitsCapacityBlock() {
+        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(), ""));
+        assertFalse(out.contains("\"capacity\":"), out);
+    }
+
     @Test
     void listReportsLeadsAndFlagsTheCallersOwnRow() {
         FakeHerdr h = new FakeHerdr();
@@ -578,7 +776,7 @@ class BridgeMcpTest {
         assertTrue(out.contains("\"role\":\"dev\""), out);
         assertTrue(out.contains("\"role\":\"reviewer\""), out);
         assertEquals(2, out.split("\"profile\":\"ltms-local\"", -1).length - 1,
-                "both members share one profile — that is the point: " + out);
+                "inert capacity is omitted, leaving the two member rows: " + out);
     }
 
     @Test

bridged/src/test/java/dev/ltms/bridged/member/ClaudeCodeLauncherTest.java

@@ -102,6 +102,53 @@ class ClaudeCodeLauncherTest {
                 "the operator's own args are preserved, in order, ahead of the model flag");
     }
 
+    @Test
+    void appendsTheBaseComposedRoleAndReplyCharter() {
+        FakeHerdr herdr = new FakeHerdr();
+        String roleCharter = "You review changes.";
+        BridgedConfig.Profile cfg = new BridgedConfig.Profile(
+                "sonnet", "http://gx00.gw:8000", null, null, "BRIDGED_WORKER_TOKEN",
+                List.of("claude"), "tab", "bridged-workers", "w #{n}",
+                "http://127.0.0.1:8765/mcp", null, null);
+        ClaudeCodeLauncher svc = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null,
+                0, 0L, () -> fleet(Map.of("reviewer", roleCharter), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
+
+        List<String> args = spawnedArgs(herdr);
+        int flag = args.indexOf("--append-system-prompt");
+        assertEquals(1, args.stream().filter("--append-system-prompt"::equals).count(),
+                "the composed charter is passed once");
+        assertEquals(roleCharter + "\n\n" + HerdrPeerLauncher.REPLY_CHARTER, args.get(flag + 1),
+                "the role charter comes first and the reply rule comes last");
+    }
+
+    @Test
+    void profileWithoutMcpOrRoleCharterGetsNoSystemPrompt() {
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of(), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.DEV));
+
+        assertFalse(spawnedArgs(herdr).contains("--append-system-prompt"));
+    }
+
+    @Test
+    void profileWithoutMcpStillGetsItsRoleCharter() {
+        FakeHerdr herdr = new FakeHerdr();
+        String roleCharter = "You design changes.";
+        ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of("architect", roleCharter), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.ARCHITECT));
+
+        List<String> args = spawnedArgs(herdr);
+        int flag = args.indexOf("--append-system-prompt");
+        assertTrue(flag >= 0, "a role charter does not need an MCP mount");
+        assertEquals(roleCharter, args.get(flag + 1));
+        assertFalse(args.contains("--mcp-config"));
+    }
+
     private ClaudeCodeLauncher multiProfile(FakeHerdr herdr) {
         BridgedConfig.Profile gx10 = new BridgedConfig.Profile("gx10", "http://gx10.gw:8000", "coder",
                 null, "BRIDGED_WORKER_TOKEN", List.of("claude"), "tab", "bridged-workers", "w #{n}", null, null, null);

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,57 @@ 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
+
+        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");
+    }
+
+    @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;
+    }
 }