fleetd #302: require content in MessageService.reply so a REST reply with no content cannot silently resolve a waiter

release() cancelled the consumer and dropped its local record of deliveries the
broker still held as outstanding. Cancelling a consumer does not requeue them:
they stay unacked on the still-open channel until it or the connection closes.
So a held-but-undrained reply became permanently unreachable — a worker's
report lost with no error and no log line. It now nacks with requeue, after
cancelling, so a later own() can still receive it.

fleetd/src/main/java/dev/ltms/fleet/Fleetd.java

@@ -625,6 +625,19 @@ public final class Fleetd {
             log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
         }
 
+        // fleetd #297: named once and reused verbatim below for FleetApp's GET /profiles, rather than
+        // built a second time — two independently-constructed sources reading the SAME BackendQuarantine
+        // / BackendOutagePolicy would still be able to drift (e.g. a future edit to the credentialIdFor
+        // closure in only one of the two places), exactly the shape #284 was.
+        FleetMcp.QuarantineSource quarantineSource = new FleetMcp.QuarantineSource(profile -> {
+            var configured = config.get().profiles().get(profile);
+            return configured == null ? null : configured.effectiveCredentialId();
+        }, quarantine);
+        FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(profile -> {
+            var configured = config.get().profiles().get(profile);
+            return configured == null ? null : configured.effectiveCredentialId();
+        }, outagePolicy);
+
         FleetMcp mcp = new FleetMcp(messages, workers, sessions, identity, presence,
                 primaryRegistry, callers, metrics, new FleetMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
                 profile -> {
@@ -636,15 +649,9 @@ public final class Fleetd {
                     return FleetHealthMonitor.coverage(health != null && health.isEnabled(),
                             health != null && health.notifications() != null && health.notifications().configured());
                 }),
-                new FleetMcp.QuarantineSource(profile -> {
-                    var configured = config.get().profiles().get(profile);
-                    return configured == null ? null : configured.effectiveCredentialId();
-                }, quarantine),
+                quarantineSource,
                 leadMailbox,
-                new FleetMcp.OutageSource(profile -> {
-                    var configured = config.get().profiles().get(profile);
-                    return configured == null ? null : configured.effectiveCredentialId();
-                }, outagePolicy),
+                outageSource,
                 new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)));
 
         // CB-637: the receive half. Only constructed when a lead mailbox actually opened — with no
@@ -715,9 +722,12 @@ public final class Fleetd {
         // GET /sessions must merge across both, or a down/unpolled member daemon is invisible.
         // fleetd #111: live (re-read-per-request) memberCredentials view for GET /member-credentials —
         // same hot-reload shape as the memberCredentials supplier passed to ClaudeCodeLauncher above.
+        // fleetd #297: quarantineSource/outageSource are the SAME instances passed to FleetMcp above —
+        // GET /profiles must report the identical quarantine/cool-off facts as fleet_profiles.
         Javalin app = new FleetApp(herdr, memberHerdr, workers, sessions, messages, presence, mcp.servlet(),
                 callers, metrics, deliverable,
-                () -> MemberCredentialPolicyView.of(config.get().memberCredentials())).build();
+                () -> MemberCredentialPolicyView.of(config.get().memberCredentials()),
+                quarantineSource, outageSource).build();
         app.start(cfg.bind().host(), cfg.bind().port());
         log.info("fleetd listening on {}:{}, herdr socket {}",
                 cfg.bind().host(), cfg.bind().port(), socket);

fleetd/src/main/java/dev/ltms/fleet/health/FleetHealthMonitor.java

@@ -14,6 +14,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
@@ -28,6 +29,14 @@ public final class FleetHealthMonitor {
     static final int MAX_FAIL_TARGET_ATTEMPTS = 3;
     // CB-641: Match the injector's 60s readiness gate so health allows a full first boot.
     static final long READINESS_GRACE_NANOS = TimeUnit.SECONDS.toNanos(60);
+    /**
+     * fleetd #280: how long after a terminal transition to wait before the one bounded re-check
+     * fires. Must exceed the worst-case reverse-rendezvous {@code fleet_ask} window (55-115s, see
+     * {@code FleetMcp.ASK_DEFAULT_TIMEOUT_MS} / {@code FleetApp.MAX_ASK_TIMEOUT_MS}) so that, if the
+     * target was genuinely {@code ASKING} when {@code state} was first observed, its own ask has had
+     * time to lapse (clearing {@code Task#question} back to {@code null}) before this fires.
+     */
+    static final long ASK_LAPSE_RECHECK_DELAY_SECONDS = 120;
 
     private final AgentControl agents;
     private final Supplier<List<MemberSession>> roster;
@@ -38,7 +47,16 @@ public final class FleetHealthMonitor {
     private final long workingSuspectAfterNanos;
     private final BiConsumer<String, String> failTarget;
     private final Map<String, HealthPrior> priors = new HashMap<>();
-    private final Map<String, HealthState> states = new HashMap<>();
+    /**
+     * The live classification per member, and the only one of this class's three maps that more
+     * than one scheduler task touches. {@code tick} writes it (and prunes it to the roster);
+     * fleetd #280's delayed {@link #recheckTerminalTarget} reads it from its own separate scheduled
+     * task. Both run on the single-threaded scheduler {@code Fleetd} passes in today, so they are
+     * serialised — but nothing in this class enforces that, and an unsynchronised {@link HashMap}
+     * read racing a resize can spin a CPU forever rather than fail visibly. {@code priors} and
+     * {@code orphanStreaks} stay plain maps because {@code tick} is still their only toucher.
+     */
+    private final Map<String, HealthState> states = new ConcurrentHashMap<>();
     /**
      * CB-643: consecutive ticks on which a target looked like an orphaned delegation. The fact
      * {@link MessageService#hasOrphanedDelegation} reports is a true snapshot, but it can read true
@@ -171,6 +189,7 @@ public final class FleetHealthMonitor {
         // member stayed terminal.
         if (terminal(next)) {
             failTerminalTarget(target, next);
+            scheduleTerminalRecheck(target, next);
         }
     }
 
@@ -191,6 +210,48 @@ public final class FleetHealthMonitor {
                 target, state, MAX_FAIL_TARGET_ATTEMPTS, last);
     }
 
+    /**
+     * fleetd #280: schedule the one bounded, delayed follow-up for a terminal transition — never a
+     * per-tick retry (CB-580 rejected that shape; {@link #reportTransition} still fires
+     * {@link #failTerminalTarget} exactly once per transition, unconditionally on the tick loop).
+     * This is a single one-shot task, scheduled once per transition into GONE/NEVER_READY, so a
+     * member stuck terminal for the rest of its life gets exactly one extra attempt, not one per
+     * tick. See {@link #recheckTerminalTarget} for why the extra attempt is safe.
+     */
+    private void scheduleTerminalRecheck(String target, HealthState state) {
+        if (scheduler.isShutdown()) return;
+        try {
+            scheduler.schedule(() -> recheckTerminalTarget(target, state),
+                    ASK_LAPSE_RECHECK_DELAY_SECONDS, TimeUnit.SECONDS);
+        } catch (RuntimeException e) {
+            log.warn("fleet health: could not schedule terminal re-check for member={} state={}",
+                    target, state, e);
+        }
+    }
+
+    /**
+     * fleetd #280: the delayed re-check {@link #scheduleTerminalRecheck} scheduled for one terminal
+     * transition. By now, a {@code fleet_ask} that was still open when {@code state} was first
+     * observed has had time to lapse on its own (see {@link #ASK_LAPSE_RECHECK_DELAY_SECONDS}),
+     * clearing {@code Task#question} back to {@code null} — which is exactly what
+     * {@link MessageService#abandon(String, String, boolean)}'s {@code sweepAsking=false} filter
+     * needs to finally match it. Calling {@link #failTerminalTarget} again is safe only because
+     * {@code sweepAsking} stays {@code false}: a task genuinely still {@code ASKING} is skipped
+     * exactly as it was on the very first attempt — this never fails a ticket whose ask has not yet
+     * lapsed.
+     *
+     * <p><strong>Guarded on "target is still classified {@code state}."</strong> Without this guard,
+     * a member that recovered (or was released and dropped from the roster) between the transition
+     * and this re-check would still take a blind {@code failTarget} call — reaching into whatever
+     * brand-new, unrelated turn it has since picked up and failing it too. {@link #states} already
+     * carries the live classification (updated every tick, pruned to the current roster on release),
+     * so a stale or recovered target simply reads as a mismatch here and this is a no-op.
+     */
+    void recheckTerminalTarget(String target, HealthState state) {
+        if (states.get(target) != state) return;
+        failTerminalTarget(target, state);
+    }
+
     private static boolean terminal(HealthState state) {
         return state == HealthState.GONE || state == HealthState.NEVER_READY;
     }

fleetd/src/main/java/dev/ltms/fleet/mcp/FleetMcp.java

@@ -1015,6 +1015,25 @@ public final class FleetMcp {
      * both maps at once when it is both exhaustion-quarantined AND cooling off.
      */
     static McpSchema.CallToolResult profiles(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
+        return text(json(profilesView(workers, quarantine, outage)));
+    }
+
+    /**
+     * The body both front doors answer {@code profiles} with: the configured profile names, the
+     * default, and the two independent outage states — {@code quarantined} (the backend reported it
+     * out of capacity) and {@code coolingOff} (the credential threw repeated non-exhaustion backend
+     * errors). Each map is present only when at least one profile is in that state, and a profile
+     * can appear in both at once, because the two checks are separate.
+     *
+     * <p>fleetd #297: extracted so {@code fleet_profiles} and {@code GET /profiles} render from ONE
+     * body builder rather than two copies. Passing both doors the same {@link QuarantineSource} and
+     * {@link OutageSource} instances is necessary but not sufficient: with the loop written out
+     * twice, a later edit to the row shape — a renamed key, an added field — lands on one door and
+     * not the other, and the two then disagree about a live outage. That is exactly what fleetd
+     * #284 was, where one rule computed in two places was widened in only one and a single response
+     * contradicted itself. Shared inputs do not make duplicated computation safe.
+     */
+    public static Map<String, Object> profilesView(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
         Map<String, Object> result = new LinkedHashMap<>();
         result.put("profiles", workers.profiles());
         result.put("default", workers.defaultProfile() == null ? "" : workers.defaultProfile());
@@ -1046,7 +1065,7 @@ public final class FleetMcp {
         if (!coolingOff.isEmpty()) {
             result.put("coolingOff", coolingOff);
         }
-        return text(json(result));
+        return result;
     }
 
     /**

fleetd/src/main/java/dev/ltms/fleet/member/HerdrPeerLauncher.java

@@ -697,7 +697,20 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
         if (paneId == null) {
             throw new IllegalStateException("pane.split returned no pane — cannot start a peer");
         }
-        Agent peer = startUniquelyNamed(cfg, argv, paneId).agent();
+        Agent peer;
+        try {
+            peer = startUniquelyNamed(cfg, argv, paneId).agent();
+        } catch (RuntimeException e) {
+            // The peer never started — don't leave the pane we just created orphaned.
+            // Best-effort cleanup; never let it mask the real spawn failure.
+            try {
+                stop(paneId);
+            } catch (RuntimeException cleanup) {
+                log.warn("failed to close orphaned pane {} after spawn error: {}",
+                        paneId, cleanup.getMessage());
+            }
+            throw e;
+        }
         log.info("{} started pane={} terminal={}", namePrefix, peer.paneId(), peer.terminalId());
         return peer;
     }
@@ -947,7 +960,14 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
             // person can go close by hand.
             try {
                 spaces.closeTab(loc.tabId());
-            } catch (HerdrException e) {
+            } catch (RuntimeException e) {
+                // Caught as RuntimeException, not HerdrException, to match releaseZdotdir's own
+                // guard five lines below. Today the two are the same set — HerdrCodec wraps every
+                // encode/decode failure and UnixSocketHerdrClient wraps every IOException, so
+                // HerdrException is all closeTab can actually throw. Narrowing to it anyway would
+                // leave this step guarded against the expected failure and bare against any other,
+                // which is the exact asymmetry fleetd #293 exists to remove. No behaviour change
+                // today; it stops a later change inside WorkspaceControl.closeTab reopening it.
                 log.warn("tab.close({}) failed — the pane is already torn down, so continuing; the "
                         + "tab may need manual cleanup: {}", loc.tabId(), e.getMessage());
             }
@@ -989,7 +1009,8 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      * that gap: it stops waiting immediately (never burns the rest of the timeout), runs the same
      * teardown the timeout path below runs, and throws with a message that says the backend exited
      * rather than that the pane was slow. Any other {@link HerdrException} still propagates
-     * unchanged — this gate does not know how to recover from it.
+     * unchanged — this gate does not interpret or recover from it, but it still closes the pane
+     * it opened before handing the exception to its caller.
      */
     private void waitUntilInjectableOrThrow(String paneId) {
         long start = nowMillis.getAsLong();
@@ -1003,7 +1024,15 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
                 if (isAlreadyGone(e)) {
                     failFastOnGoneBackend(paneId, e, nowMillis.getAsLong() - start);
                 }
-                throw e; // any other herdr failure is not ours to interpret — let it propagate
+                // This gate must not interpret an unrelated herdr error, but the caller does not
+                // receive paneId when spawn throws. Close the pane here before propagating e unchanged.
+                try {
+                    stop(paneId);
+                } catch (RuntimeException cleanup) {
+                    log.warn("failed to close orphaned pane {} after readiness-gate error: {}",
+                            paneId, cleanup.getMessage());
+                }
+                throw e;
             }
             lastStatus = sample.status();
             if (lastStatus.injectable() || refinedInjectable(paneId, sample)) {

fleetd/src/main/java/dev/ltms/fleet/msg/AmqpReplyInbox.java

@@ -208,18 +208,72 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
         }
     }
 
+    /**
+     * Release ownership of {@code target}: cancel its consumer, then nack-with-requeue every
+     * delivery still held for it instead of just dropping the local record.
+     *
+     * <p><strong>Cancelling a consumer does not requeue its in-flight deliveries.</strong> In AMQP,
+     * a delivery that was pushed to a consumer stays unacked, attached to the still-open
+     * {@link #channel}, until that channel or the connection closes — {@code basicCancel} alone does
+     * neither. So before this method existed with a requeue step, it dropped {@link #held}'s entries
+     * for {@code target} while the broker still considered them outstanding: never acked, never
+     * nacked, never requeued, and no longer reachable by {@link #peek} — permanently invisible. This
+     * is unlike {@link #handleRecovery} and {@link #close()}, whose bare {@code held.clear()} is
+     * correct because each has already made the broker requeue (a real connection drop, or
+     * {@code channel.close()} respectively) before clearing local state.
+     *
+     * <p><strong>Order: cancel first, then nack.</strong> A delivery tag stays valid for
+     * {@code basicNack} on this channel regardless of whether its consumer is still attached — only
+     * a channel/connection close invalidates it — so cancelling {@code target}'s consumer first does
+     * not risk the tags. Doing it the other way round does: nacking a delivery with {@code requeue}
+     * while its consumer is still active hands the message straight back to that <em>same</em>
+     * consumer the instant a prefetch slot frees up (confirmed against a real broker — see
+     * {@code AmqpReplyInboxContractTest.releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery}),
+     * which races this method's own {@code held.remove(target)}: the redelivery can land after the
+     * clear and leave a stale entry behind, so {@link #peek} is no longer reliably empty right after
+     * {@link #release}. Cancelling first closes that consumer, so the requeued message goes back to
+     * the queue for whichever consumer picks it up next (a later {@link #own}), not this one.
+     *
+     * <p><strong>Failure of the requeue is best-effort, not fatal.</strong> {@link #release} runs
+     * during teardown ({@code Fleetd} calls it right after {@code MessageService.abandon}), and a
+     * throw here would abort cleanups the caller depends on — the same argument fleetd #293 settled
+     * for {@code HerdrPeerLauncher.stop()}'s tab-close step. So a failed {@code basicNack} is logged
+     * at WARN, naming the target and delivery tag that leaked, and release proceeds; the delivery
+     * stays unacked on the broker rather than being silently dropped, so it is still recoverable by a
+     * later connection drop even though this release did not manage to requeue it immediately. A
+     * failed {@code basicCancel} still throws, unchanged from before this fix — that failure means
+     * the consumer may still be attached, so best-effort requeue is not attempted underneath it.
+     */
     @Override
     public void release(String target) {
         synchronized (channelLock) {
             String tag = consumerTags.remove(target);
-            held.remove(target);                   // stale delivery tags must not survive release
-            if (tag == null) {
-                return;
+            if (tag != null) {
+                try {
+                    channel.basicCancel(tag);
+                } catch (IOException e) {
+                    throw new IllegalStateException("cannot cancel consumer for " + target, e);
+                }
             }
-            try {
-                channel.basicCancel(tag);
-            } catch (IOException e) {
-                throw new IllegalStateException("cannot cancel consumer for " + target, e);
+            var perTarget = held.remove(target);
+            if (perTarget != null) {
+                synchronized (perTarget) {
+                    for (Held h : perTarget.values()) {
+                        try {
+                            channel.basicNack(h.deliveryTag(), false, true); // requeue, don't drop
+                        } catch (IOException | RuntimeException e) {
+                            // Caught broadly (not just IOException) for the same reason #293 catches
+                            // RuntimeException in HerdrPeerLauncher.stop(): best-effort teardown must
+                            // not be guarded only against the expected failure and bare against any
+                            // other. The message stays unacked on the broker either way — not lost,
+                            // just not proactively requeued — until a connection drop frees it.
+                            log.warn("release({}): could not requeue held delivery (msgId={}, tag={})"
+                                            + " back to the broker — it stays unacked until a connection"
+                                            + " drop frees it: {}",
+                                    target, h.message().msgId(), h.deliveryTag(), e.getMessage());
+                        }
+                    }
+                }
             }
         }
     }

fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java

@@ -408,10 +408,28 @@ public final class MessageService {
      * as the zero-candidate case does, and let {@link #abandon} apply the eventual recovery
      * deterministically instead.
      *
+     * <p><strong>{@code content} is required (fleetd #302).</strong> Both doors that reach this
+     * method must reject a missing/blank reply the same way, so the check lives here rather than in
+     * either caller: {@code FleetMcp.reply} already refuses a {@code null} content before it ever
+     * calls this method (its own required-arg guard), and no test or production call site anywhere
+     * in the codebase relies on replying with empty content — confirmed by searching every call site
+     * of this method before adding the check, not assumed. Without this guard, a REST {@code
+     * POST /sessions/{id}/reply} whose body omits {@code content} (or a client library that maps a
+     * missing field to {@code ""}) used to reach {@link Rendezvous#resolve} with an empty string,
+     * silently completing the lead's blocking wait with nothing — indistinguishable from a worker
+     * that genuinely replied with nothing, which is worse than a loud failure because it destroys the
+     * information that the reply never arrived.
+     *
+     * @throws IllegalArgumentException if {@code content} is {@code null} or blank — the caller must
+     *         report this as a client error (REST: 400 {@code bad_request}) rather than resolve
+     *         anything
      * @return always {@code true} — the reply resolved a live send, completed a parked ticket, or
      *         was queued
      */
     public boolean reply(String session, String content) {
+        if (content == null || content.isBlank()) {
+            throw new IllegalArgumentException("content is required");
+        }
         if (rendezvous.resolve(session, content)) {
             count(FleetMetrics.REPLIES, "path", "rendezvous");
             return true;                       // a live send took it — unchanged fast path

fleetd/src/main/java/dev/ltms/fleet/rest/FleetApp.java

@@ -9,6 +9,7 @@ import dev.ltms.fleet.auth.Principal;
 import dev.ltms.fleet.guard.GuardException;
 import dev.ltms.fleet.metrics.FleetMetrics;
 import dev.ltms.fleet.metrics.Metrics;
+import dev.ltms.fleet.mcp.FleetMcp;
 import dev.ltms.fleet.herdr.Agent;
 import dev.ltms.fleet.herdr.HerdrClient;
 import dev.ltms.fleet.herdr.HerdrException;
@@ -86,6 +87,12 @@ public final class FleetApp {
     // absent() (the honest "no policy configured" view) for every constructor that does not wire
     // a real one, so existing legacy call sites keep building without knowing this field exists.
     private final Supplier<MemberCredentialPolicyView> memberCredentials;
+    // fleetd #297: the SAME shared sources FleetMcp.profiles/fleet_profiles reads (BackendQuarantine
+    // and BackendOutagePolicy are each one instance for the whole daemon — see Fleetd wiring) so
+    // GET /profiles cannot drift from fleet_profiles about which profile is quarantined/cooling off.
+    // .none() (the honest "feature not wired" view) for every constructor that does not pass one.
+    private final FleetMcp.QuarantineSource quarantine;
+    private final FleetMcp.OutageSource outage;
     private final ObjectMapper mapper = new ObjectMapper();
 
     /**
@@ -146,6 +153,23 @@ public final class FleetApp {
                        MessageService messages, MemberPresence presence,
                        HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
                        Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials) {
+        this(herdr, memberHerdr, workers, sessions, messages, presence, mcpServlet, auth, metrics,
+                deliverable, memberCredentials, FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
+    }
+
+    /**
+     * @param quarantine the SAME {@link FleetMcp.QuarantineSource} instance passed to {@code
+     *                   FleetMcp} (fleetd #297), so {@code GET /profiles} reports the identical
+     *                   exhaustion-quarantine facts as {@code fleet_profiles} rather than a second,
+     *                   independently-computed copy
+     * @param outage     the SAME {@link FleetMcp.OutageSource} instance passed to {@code FleetMcp} —
+     *                   see {@code quarantine}; a SEPARATE check from it, never merged in
+     */
+    public FleetApp(HerdrClient herdr, HerdrClient memberHerdr, PeerLauncher workers, SessionManager sessions,
+                       MessageService messages, MemberPresence presence,
+                       HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
+                       Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials,
+                       FleetMcp.QuarantineSource quarantine, FleetMcp.OutageSource outage) {
         this.herdr = herdr;
         this.memberHerdr = memberHerdr != null ? memberHerdr : herdr;
         this.workers = workers;
@@ -156,6 +180,8 @@ public final class FleetApp {
         this.auth = auth;
         this.metrics = metrics;
         this.memberCredentials = memberCredentials != null ? memberCredentials : MemberCredentialPolicyView::absent;
+        this.quarantine = quarantine != null ? quarantine : FleetMcp.QuarantineSource.none();
+        this.outage = outage != null ? outage : FleetMcp.OutageSource.none();
     }
 
     /** Wire routes onto a fresh, unstarted Javalin instance. Caller starts it. */
@@ -338,8 +364,15 @@ public final class FleetApp {
         if (!allow(ctx, routeAction("GET /agents"), null)) {
             return;
         }
-        ctx.status(200).json(Map.of("agents",
-                workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
+        try {
+            ctx.status(200).json(Map.of("agents",
+                    workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
+        } catch (HerdrException e) {
+            // fleetd #297: workers.list() reaches herdr — a transport failure must land in the same
+            // {error, detail} envelope every other failure path here uses, not escape as a bare
+            // exception and leave Javalin's default handling to respond outside the JSON contract.
+            herdrError(ctx, e);
+        }
     }
 
     /** CB-304: bridge-owned roster merged with live herdr status by paneId. */
@@ -347,40 +380,57 @@ public final class FleetApp {
         if (!allow(ctx, routeAction("GET /members"), null)) {
             return;
         }
-        // CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
-        Map<String, Agent> live = workers.list().stream()
-                .map(Agent.class::cast)
-                .filter(a -> a.terminalId() != null)
-                .collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
-        // fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
-        // uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
-        List<Map<String, Object>> out = sessions.rosterResolved().stream()
-                .map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
-                .toList();
-        Map<String, Object> body = new LinkedHashMap<>();
-        // fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
-        // "workers", so a caller that read "members" saw an empty fleet and reported no members at
-        // all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
-        // REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
-        // drops reads this endpoint. Drop the alias once nothing reads it.
-        body.put("members", out);
-        body.put("workers", out);
-        // CB-586: operator visibility for the refs/wip snapshot store without shelling into the
-        // repo — how many snapshot refs exist and roughly what they cost. Present only once a
-        // worktree session has established the repo, so a never-snapshotted fleet reports nothing.
-        sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
-                Map.of("count", st.count(), "costBytes", st.costBytes())));
-        ctx.status(200).json(body);
+        try {
+            // CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
+            Map<String, Agent> live = workers.list().stream()
+                    .map(Agent.class::cast)
+                    .filter(a -> a.terminalId() != null)
+                    .collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
+            // fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
+            // uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
+            List<Map<String, Object>> out = sessions.rosterResolved().stream()
+                    .map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
+                    .toList();
+            Map<String, Object> body = new LinkedHashMap<>();
+            // fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
+            // "workers", so a caller that read "members" saw an empty fleet and reported no members at
+            // all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
+            // REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
+            // drops reads this endpoint. Drop the alias once nothing reads it.
+            body.put("members", out);
+            body.put("workers", out);
+            // CB-586: operator visibility for the refs/wip snapshot store without shelling into the
+            // repo — how many snapshot refs exist and roughly what they cost. Present only once a
+            // worktree session has established the repo, so a never-snapshotted fleet reports nothing.
+            sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
+                    Map.of("count", st.count(), "costBytes", st.costBytes())));
+            ctx.status(200).json(body);
+        } catch (HerdrException e) {
+            // fleetd #297: same reasoning as agents() above — this is the out-of-band roster a lead
+            // falls back to when its MCP mount drops, so it must stay inside the JSON error contract
+            // exactly when herdr is briefly unreachable, not escape as a bare exception.
+            herdrError(ctx, e);
+        }
     }
 
-    /** The configured worker profiles and which one a no-argument spawn uses. */
+    /**
+     * The configured worker profiles, which one a no-argument spawn uses, and (fleetd #297) the two
+     * outage states {@code fleet_profiles} already reports: {@code quarantined} (CB-578 stage B —
+     * the backend reported it out of capacity) and {@code coolingOff} (fleetd #201 Unit 5 — the
+     * credential threw repeated non-exhaustion backend errors). Both are read from the SAME shared
+     * {@link FleetMcp.QuarantineSource}/{@link FleetMcp.OutageSource} instances {@code FleetMcp}
+     * reads, never recomputed, so the two doors cannot disagree about which profile is down and why.
+     * Independent checks, so a profile can appear in both maps at once; each map is present only
+     * when at least one profile is in that state.
+     */
     private void profiles(Context ctx) {
         if (!allow(ctx, routeAction("GET /profiles"), null)) {
             return;
         }
-        ctx.status(200).json(Map.of(
-                "profiles", workers.profiles(),
-                "default", workers.defaultProfile() == null ? "" : workers.defaultProfile()));
+        // fleetd #297: ONE body builder, shared with fleet_profiles. Handing both doors the same
+        // QuarantineSource/OutageSource instances stops them reading different facts; rendering
+        // through the same method stops them reporting those facts differently. Both are needed.
+        ctx.status(200).json(FleetMcp.profilesView(workers, quarantine, outage));
     }
 
     /**
@@ -634,7 +684,19 @@ public final class FleetApp {
             ctx.status(400).json(Map.of("error", "bad_request", "detail", "body must be JSON"));
             return;
         }
-        messages.reply(id, content);
+        // fleetd #302: content is required. `.path("content").asText("")` above turns a missing key
+        // into "" rather than throwing, so without this check an empty/blank reply used to reach
+        // messages.reply(...) and silently resolve the lead's waiter — the same class of bug as the
+        // sibling "content is required" guards on sendMessage/askMessage below, except this one wrote
+        // a WRONG value instead of failing loudly. The check lives in MessageService.reply so both
+        // this door and FleetMcp.reply inherit the same rule; this catch only translates it into the
+        // {error, detail} envelope this file uses everywhere else.
+        try {
+            messages.reply(id, content);
+        } catch (IllegalArgumentException e) {
+            ctx.status(400).json(Map.of("error", "bad_request", "detail", e.getMessage()));
+            return;
+        }
         ctx.status(200).json(Map.of("sessionId", id, "delivered", true));
     }
 

fleetd/src/main/java/dev/ltms/fleet/session/SessionManager.java

@@ -864,10 +864,20 @@ public final class SessionManager implements TurnListener {
     }
 
     /**
-     * Gracefully drain all registered sessions on daemon shutdown. For each session that is
-     * {@code BUSY}, poll up to {@code timeoutNanos} for it to leave {@code BUSY}, then release it
-     * regardless. Non-busy sessions are released immediately. A failure releasing one session is
-     * logged and does not abort the rest.
+     * Gracefully drain all registered sessions on daemon shutdown. Non-busy sessions are released
+     * immediately; a {@code BUSY} one is polled until it leaves {@code BUSY}, then released
+     * regardless. A failure releasing one session is logged and does not abort the rest.
+     *
+     * <p>{@code timeoutNanos} is a budget for the WHOLE drain, not a grace period per session: the
+     * deadline is taken once, before the loop. So the first BUSY session can spend all of it, and a
+     * later BUSY one is then released with no wait at all. That is deliberate. This drain is only
+     * one phase of shutdown — {@code Fleetd} closes the message service, the push loop, the
+     * heartbeat, MCP and the router after it — and the whole sequence has to finish inside
+     * launchd's exit window. A per-session grace would let N busy members drain for N * the
+     * timeout, overrun that window, and get the daemon SIGKILLed part-way through; the members not
+     * yet reached would then get no clean release, no preserved-worktree log, and no snapshot.
+     * Cutting one turn short is the cheaper failure, and it is not silent: an abandoned BUSY
+     * session is preserved, snapshotted, and logged at WARN by {@code logPreservedForShutdown}.
      *
      * <p>CB-544: this is a {@link ReleaseCause#SHUTDOWN} release — the worker's pane is stopped
      * (the process must end) but its worktree is preserved and its path logged. Shutdown is never

fleetd/src/test/java/dev/ltms/fleet/health/FleetHealthMonitorTest.java

@@ -24,7 +24,9 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -434,4 +436,120 @@ class FleetHealthMonitorTest {
             logger.detachAppender(appender);
         }
     }
+
+    // --- fleetd #280: a GONE/NEVER_READY guess whose fleet_ask lapses AFTER the first sweep must
+    // still be swept, without ever reaching into a target that has since recovered or left the
+    // roster. See FleetHealthMonitor.recheckTerminalTarget's javadoc for the full reachability chain.
+
+    @Test void terminalTransitionSchedulesExactlyOneDelayedRecheck() {
+        ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1);
+        FleetHealthMonitor monitor = monitor(new FakeHerdr(),
+                List.of(member("term_a", MemberSession.State.BUSY, 0, 0)), scheduler, () -> 1, 600,
+                (_, _) -> { });
+
+        monitor.reportTransition("term_a", HealthState.GONE);
+
+        // Driven via reportTransition directly (not tick()), so the queue holds only the recheck.
+        assertEquals(1, scheduler.getQueue().size());
+        ScheduledFuture<?> scheduled = (ScheduledFuture<?>) scheduler.getQueue().peek();
+        assertTrue(scheduled.getDelay(TimeUnit.SECONDS) > 100,
+                "the delay must clear the worst-case fleet_ask lapse window (up to 115s)");
+
+        // An unchanged tick must not queue a second one (CB-580's fire-once rule extends to this).
+        monitor.reportTransition("term_a", HealthState.GONE);
+        assertEquals(1, scheduler.getQueue().size());
+        monitor.stop();
+    }
+
+    @Test void recheckIsANoOpOnceTheTargetHasRecovered() {
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        assertEquals(1, failTarget.calls.size());
+
+        monitor.reportTransition("term_a", HealthState.IDLE); // recovered before the recheck fired
+        monitor.recheckTerminalTarget("term_a", HealthState.GONE);
+
+        assertEquals(1, failTarget.calls.size(), "a recovered target must not be reached into again");
+        monitor.stop();
+    }
+
+    @Test void recheckIsANoOpForATargetItNeverObserved() {
+        // Mirrors "left the roster": tick() prunes states.keySet() to the current roster on release
+        // (see FleetHealthMonitor.tick), so a target this monitor never recorded is the same case.
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+
+        monitor.recheckTerminalTarget("term_never_seen", HealthState.GONE);
+
+        assertEquals(0, failTarget.calls.size(), "an untracked/released target must not be reached into");
+        monitor.stop();
+    }
+
+    /**
+     * The scenario from the ticket, end to end, driven through the real {@link MessageService}: a
+     * target's ask is still genuinely open when health first observes GONE (sweep must skip it,
+     * exactly as {@code abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer} pins), the ask then lapses
+     * on its own, an unchanged tick still must not refire, and only the delayed recheck sweeps the
+     * now-lapsed ticket to FAILED.
+     */
+    @Test void delayedRecheckSweepsATicketWhoseAskLapsedAfterGoneWasFirstObserved() throws Exception {
+        FakeHerdr herdr = new FakeHerdr().withAgent("worker", "term_a", "pane-term_a", "tab_a")
+                .readText("$ prompt");
+        AgentControl agents = new AgentControl(herdr);
+        Rendezvous rendezvous = new Rendezvous();
+        Injector injector = new Injector(agents);
+        InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+        inbox.own("term_a");
+        MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents,
+                () -> List.of(member("term_a", MemberSession.State.READY, 0, 0)), messages,
+                new ScheduledThreadPoolExecutor(1), () -> 1, 60, 600, messages::abandon);
+
+        String ticket = messages.sendAsync("term_a", "task that asks");
+        long deadline = System.currentTimeMillis() + 2000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"), "the async send should have opened its waiter");
+        injector.onStatus("term_a", AgentStatus.IDLE);    // deliver the task
+        injector.onStatus("term_a", AgentStatus.WORKING); // the worker picks it up
+
+        // The worker asks, with a short timeout so its own fleet_ask lapses quickly in test time.
+        CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
+                () -> messages.ask("term_a", "which config?", 200));
+        awaitPhase(messages, ticket, MessageService.Phase.ASKING);
+
+        monitor.reportTransition("term_a", HealthState.GONE);
+        assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase(),
+                "the first sweep must not fail a ticket that is still genuinely being asked");
+
+        // The worker's own fleet_ask now lapses on its own — task.question clears to null.
+        assertEquals(MessageService.AskOutcome.TIMED_OUT, ask.get(5, TimeUnit.SECONDS).outcome());
+
+        // An unchanged tick still must not refire (CB-580).
+        monitor.reportTransition("term_a", HealthState.GONE);
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase());
+
+        // The delayed recheck scheduled for the original transition finally sweeps it.
+        monitor.recheckTerminalTarget("term_a", HealthState.GONE);
+        assertEquals(MessageService.Phase.FAILED, messages.poll(ticket).phase());
+
+        monitor.stop();
+    }
+
+    private static MessageService.TaskView awaitPhase(MessageService messages, String ticket,
+            MessageService.Phase phase) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + 2000;
+        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;
+    }
 }

fleetd/src/test/java/dev/ltms/fleet/herdr/FakeHerdr.java

@@ -40,6 +40,7 @@ public final class FakeHerdr implements HerdrClient {
     private final Map<String, List<String>> extraTabs = new LinkedHashMap<>();
     private int agentNameTakenFor = 0;
     private int agentPaneBusyFor = 0;
+    private String agentStartErrorCode = null;
     private int workerTabPaneCount = 1;
     private String paneCloseErrorCode = null;
     private final Map<String, String> paneCloseErrorCodeFor = new ConcurrentHashMap<>();
@@ -84,6 +85,12 @@ public final class FakeHerdr implements HerdrClient {
         return this;
     }
 
+    /** Make every {@code agent.start} call fail with this herdr error code. */
+    public FakeHerdr agentStartFailsWith(String code) {
+        this.agentStartErrorCode = code;
+        return this;
+    }
+
     /** Make the worker tab (w9:t2) report this many panes in {@code tab.list} (default 1). */
     public FakeHerdr withWorkerTabPaneCount(int n) {
         this.workerTabPaneCount = n;
@@ -311,6 +318,10 @@ public final class FakeHerdr implements HerdrClient {
                                             + required + "`", "invalid_request", null);
                         }
                     }
+                    if (agentStartErrorCode != null) {
+                        throw new HerdrException("herdr error [" + agentStartErrorCode + "]: agent.start failed",
+                                agentStartErrorCode, null);
+                    }
                     long starts = calls.stream().filter(c -> c.method().equals("agent.start")).count();
                     if (starts <= agentPaneBusyFor) {
                         throw new HerdrException(

fleetd/src/test/java/dev/ltms/fleet/member/ClaudeCodeLauncherTest.java

@@ -1165,7 +1165,8 @@ class ClaudeCodeLauncherTest {
     @Test
     void spawnLetsAnUnrelatedHerdrErrorPropagateUnchanged() {
         // Fix 1 must only special-case a "*_not_found" answer. Any other herdr failure keeps
-        // propagating as-is — this gate does not know how to recover from it.
+        // propagating as-is — this gate does not know how to recover from it. The pane still needs
+        // closing because spawn throws before it can return the pane id to a caller that could stop it.
         FakeHerdr herdr = new FakeHerdr();
         herdr.agentStatus("unknown");
         herdr.agentGetFailsWithAfter(0, "internal_error");
@@ -1182,8 +1183,27 @@ class ClaudeCodeLauncherTest {
                 () -> svc.spawn(new SpawnRequest(null, null, null)));
 
         assertEquals("internal_error", ex.code());
-        assertEquals(0, paneCloseCount(herdr, "w9:pRoot_1"),
-                "an error this gate does not recognize is not this gate's teardown to run");
+        assertEquals(1, paneCloseCount(herdr, "w9:pRoot_1"),
+                "the unchanged error leaves spawn without a pane id, so this gate closes its orphaned pane");
+    }
+
+    @Test
+    void panePlacementClosesTheSplitPaneWhenAgentStartFails() {
+        FakeHerdr herdr = new FakeHerdr().agentStartFailsWith("internal_error");
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
+                List.of("claude"), "pane", "fleetd-workers", "w #{n}", null, 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);
+
+        dev.ltms.fleet.herdr.HerdrException ex = assertThrows(
+                dev.ltms.fleet.herdr.HerdrException.class,
+                () -> svc.spawn(new SpawnRequest(null, null, null)));
+
+        assertEquals("internal_error", ex.code(), "agent.start failure propagates unchanged");
+        assertEquals(1, paneCloseCount(herdr, "w1:pSplit"),
+                "the pane split for a peer that never starts is closed instead of left orphaned");
     }
 
     // --- fleetd #176 fix 2: corroborated UNKNOWN refinement --------------------------------------

fleetd/src/test/java/dev/ltms/fleet/msg/AmqpReplyInboxContractTest.java

@@ -154,7 +154,7 @@ class AmqpReplyInboxContractTest {
     }
 
     @Test
-    void releaseCancelsConsumerAndClearsHeld() throws Exception {
+    void releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery() throws Exception {
         String target = "worker-release-" + System.nanoTime();
         try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
             inbox.own(target);
@@ -164,6 +164,22 @@ class AmqpReplyInboxContractTest {
             inbox.release(target);
             assertTrue(inbox.peek(target).isEmpty(),
                     "release clears the local held snapshot");
+
+            // fleetd #298: release() must not just drop the local record — the broker delivery was
+            // never acked, so cancelling the consumer alone leaves it unacked-but-orphaned on the
+            // still-open channel unless release() nacks it back with requeue=true. Prove the message
+            // is genuinely recoverable, not merely absent from peek: re-own the same target and
+            // confirm the broker redelivers it to the fresh consumer.
+            inbox.own(target);
+            List<ReplyInbox.InboxMessage> recovered = awaitPeek(inbox, target);
+            assertEquals(1, recovered.size(),
+                    "a reply held (but undrained) at release() time must still be recoverable — "
+                            + "release() must requeue it, not silently drop it while the broker still "
+                            + "considers it outstanding");
+            assertEquals("m1", recovered.getFirst().msgId());
+            assertEquals("release me", recovered.getFirst().content());
+
+            inbox.ack(target, "m1");
         }
     }
 

fleetd/src/test/java/dev/ltms/fleet/rest/FleetAppTest.java

@@ -19,6 +19,10 @@ import dev.ltms.fleet.session.SessionManager;
 import dev.ltms.fleet.session.Worktrees;
 import dev.ltms.fleet.member.ClaudeCodeLauncher;
 import dev.ltms.fleet.member.CompositePeerLauncher;
+import dev.ltms.fleet.member.MemberCredentialPolicyView;
+import dev.ltms.fleet.mcp.FleetMcp;
+import dev.ltms.fleet.placement.BackendOutagePolicy;
+import dev.ltms.fleet.placement.BackendQuarantine;
 import dev.ltms.fleet.placement.PlacementPolicies;
 import io.javalin.Javalin;
 import org.junit.jupiter.api.AfterEach;
@@ -32,6 +36,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -70,6 +75,19 @@ class FleetAppTest {
 
     private int start(FakeHerdr herdr, String workerBaseUrl, Set<String> allow, String placement,
                       Worktrees worktrees, Predicate<String> deliverable) {
+        return start(herdr, workerBaseUrl, allow, placement, worktrees, deliverable,
+                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
+    }
+
+    /**
+     * fleetd #297: same wiring as above, plus the two SAME shared sources {@code GET /profiles}
+     * must read — lets a test prove the quarantined/coolingOff facts it reports come from a real
+     * {@link dev.ltms.fleet.placement.BackendQuarantine}/{@link
+     * dev.ltms.fleet.placement.BackendOutagePolicy}, exactly like {@code fleet_profiles}'s own tests.
+     */
+    private int start(FakeHerdr herdr, String workerBaseUrl, Set<String> allow, String placement,
+                      Worktrees worktrees, Predicate<String> deliverable,
+                      FleetMcp.QuarantineSource quarantine, FleetMcp.OutageSource outage) {
         FleetConfig.Profile wcfg = new FleetConfig.Profile(
                 "ltms-local", workerBaseUrl, "coder", null, "FLEETD_WORKER_TOKEN", null,
                 placement, "fleet", "worker: {profile} #{n}", null, null, null);
@@ -90,8 +108,9 @@ class FleetAppTest {
         // it directly so the inbox contract holds for those endpoints.
         inbox.own("term_a");
         MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
-        app = new FleetApp(herdr, workers, sessions, messages, this.presence, null,
-                null, null, id -> this.presence.isPresent(id) || deliverable.test(id))
+        app = new FleetApp(herdr, herdr, workers, sessions, messages, this.presence, null,
+                null, null, id -> this.presence.isPresent(id) || deliverable.test(id),
+                MemberCredentialPolicyView::absent, quarantine, outage)
                 .build().start("127.0.0.1", 0);
         return app.port();
     }
@@ -164,6 +183,22 @@ class FleetAppTest {
         assertEquals("idle", agents.get(0).get("status").asText());
     }
 
+    /**
+     * fleetd #297 gap 1: {@code workers.list()} reaches herdr, and a transport failure there must
+     * land in the same {@code {error, detail}} envelope every other failure path in this file uses
+     * (see {@code herdrError}), not escape as a bare exception outside the JSON contract.
+     */
+    @Test
+    void agentsMapsAHerdrFailureToTheJsonErrorEnvelope() throws Exception {
+        FakeHerdr down = new FakeHerdr().healthy(false);
+        int port = start(down, "http://gx00.gw:8000", Set.of("gx00.gw"));
+        HttpResponse<String> res = req(port, "GET", "/agents");
+        assertEquals(502, res.statusCode(), res.body());
+        JsonNode body = mapper.readTree(res.body());
+        assertEquals("herdr_error", body.get("error").asText());
+        assertTrue(body.has("detail"), res.body());
+    }
+
     @Test
     void spawnWorkerLandsInOwnTabInWorkerSpaceAndInjectsBaseUrl() throws Exception {
         FakeHerdr herdr = new FakeHerdr();
@@ -203,6 +238,42 @@ class FleetAppTest {
         JsonNode body = mapper.readTree(req(port, "GET", "/profiles").body());
         assertEquals("ltms-local", body.get("default").asText());
         assertEquals("ltms-local", body.get("profiles").get(0).asText());
+        assertFalse(body.has("quarantined"), "nothing is quarantined, so the key is omitted: " + body);
+        assertFalse(body.has("coolingOff"), "nothing is cooling off, so the key is omitted: " + body);
+    }
+
+    /**
+     * fleetd #297 gap 2: {@code GET /profiles} must report the same two outage states {@code
+     * fleet_profiles} does — CB-578 stage B exhaustion quarantine and fleetd #201 Unit 5 cool-off —
+     * reading the SAME shared {@link BackendQuarantine}/{@link BackendOutagePolicy} instances rather
+     * than recomputing them. The two checks are independent, and this profile is deliberately put in
+     * both states at once, matching {@code FleetMcpTest}'s own coverage of that overlap.
+     */
+    @Test
+    void profilesReportsQuarantineAndCoolingOffFromTheSameSharedSources() throws Exception {
+        FakeHerdr herdr = new FakeHerdr();
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        FleetMcp.QuarantineSource quarantineSource = new FleetMcp.QuarantineSource(
+                profile -> "ltms-local".equals(profile) ? "shared-openai" : null, quarantine);
+        BackendOutagePolicy outagePolicy = new BackendOutagePolicy(() -> 0L);
+        outagePolicy.record("shared-openai", "t1", "API Error: rate limited");
+        outagePolicy.record("shared-openai", "t2", "API Error: rate limited"); // 2nd distinct target starts the incident
+        FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(
+                profile -> "ltms-local".equals(profile) ? "shared-openai" : null, outagePolicy);
+        int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"), "tab", new GitWorktrees(),
+                ignored -> false, quarantineSource, outageSource);
+
+        JsonNode body = mapper.readTree(req(port, "GET", "/profiles").body());
+        assertTrue(body.has("quarantined"), body.toString());
+        assertEquals("shared-openai",
+                body.get("quarantined").get("ltms-local").get("credentialId").asText());
+        assertEquals(1800,
+                body.get("quarantined").get("ltms-local").get("quarantinedForSeconds").asLong());
+        assertTrue(body.has("coolingOff"), body.toString());
+        assertEquals("shared-openai",
+                body.get("coolingOff").get("ltms-local").get("credentialId").asText());
+        assertEquals(60, body.get("coolingOff").get("ltms-local").get("coolingOffForSeconds").asLong());
     }
 
     @Test
@@ -239,6 +310,23 @@ class FleetAppTest {
                 "liveStatus is unknown when herdr has no matching pane");
     }
 
+    /**
+     * fleetd #297 gap 1: same reasoning as {@code agentsMapsAHerdrFailureToTheJsonErrorEnvelope} —
+     * {@code GET /members} is the endpoint's own comment names as "the out-of-band path a lead falls
+     * back to when its MCP mount drops", so it must stay inside the {@code {error, detail}} envelope
+     * exactly when herdr is briefly unreachable.
+     */
+    @Test
+    void membersMapsAHerdrFailureToTheJsonErrorEnvelope() throws Exception {
+        FakeHerdr down = new FakeHerdr().healthy(false);
+        int port = start(down, "http://gx00.gw:8000", Set.of("gx00.gw"));
+        HttpResponse<String> res = req(port, "GET", "/members");
+        assertEquals(502, res.statusCode(), res.body());
+        JsonNode body = mapper.readTree(res.body());
+        assertEquals("herdr_error", body.get("error").asText());
+        assertTrue(body.has("detail"), res.body());
+    }
+
     @Test
     void spawnWithACwdParamRootsTheWorkerThere() throws Exception {
         FakeHerdr herdr = new FakeHerdr();
@@ -444,6 +532,53 @@ class FleetAppTest {
         assertEquals("orphan", body.get("replies").get(0).get("content").asText());
     }
 
+    @Test
+    void replyWithMissingContentIsRejectedAndDoesNotResolveTheWaiter() throws Exception {
+        // fleetd #302: `.path("content").asText("")` used to turn a missing "content" key into an
+        // empty string that reached rendezvous.resolve, silently completing the lead's blocking wait
+        // with nothing. Prove the fix two ways: the bad call is rejected with 400, AND the real send
+        // it would have wrongly resolved is still open afterwards — a real reply completes it.
+        FakeHerdr herdr = new FakeHerdr().agentStatus("idle");
+        int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
+
+        var send = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
+            try { return postMessage(port, "{\"content\":\"review this\",\"timeoutMs\":4000}"); }
+            catch (Exception e) { throw new RuntimeException(e); }
+        });
+
+        Thread.sleep(200); // let the background send open its rendezvous waiter
+
+        HttpResponse<String> badReply = postJson(port, "/sessions/term_a/reply", "{}");
+        assertEquals(400, badReply.statusCode());
+        JsonNode err = mapper.readTree(badReply.body());
+        assertEquals("bad_request", err.get("error").asText());
+        assertTrue(err.has("detail"));
+
+        // The waiter must still be open — a real reply now completes the ORIGINAL send.
+        HttpResponse<String> goodReply = postJson(port, "/sessions/term_a/reply", "{\"content\":\"LGTM ship it\"}");
+        assertEquals(200, goodReply.statusCode());
+
+        HttpResponse<String> res = send.get(6, java.util.concurrent.TimeUnit.SECONDS);
+        assertEquals(200, res.statusCode());
+        assertEquals("LGTM ship it", mapper.readTree(res.body()).get("reply").asText());
+    }
+
+    @Test
+    void replyWithEmptyOrWhitespaceContentIsRejectedSameAsMissing() throws Exception {
+        // fleetd #302 sibling case: present-but-blank content is treated the same as a missing key —
+        // FleetMcp's own required-content guard (fleet_reply's "content is required") makes no
+        // distinction between the two either, so diverging here would be a new asymmetry.
+        int port = startHealthy();
+
+        HttpResponse<String> empty = postJson(port, "/sessions/term_a/reply", "{\"content\":\"\"}");
+        assertEquals(400, empty.statusCode());
+        assertEquals("bad_request", mapper.readTree(empty.body()).get("error").asText());
+
+        HttpResponse<String> whitespace = postJson(port, "/sessions/term_a/reply", "{\"content\":\"   \"}");
+        assertEquals(400, whitespace.statusCode());
+        assertEquals("bad_request", mapper.readTree(whitespace.body()).get("error").asText());
+    }
+
     @Test
     void drainRepliesReturnsEmptyForNoReplies() throws Exception {
         int port = startHealthy();