fleetd #446 round 3: extract exhaustionSink and pin its caller (Cell A/B)

Round 2 pinned usageLimitFixWarning/usageLimitFixWarningNoModel's TEXT via
FleetdUsageLimitFixWarningTest, but a mutation battery against the merged PR
proved two gaps in the caller that builds main()'s real quarantine
ExhaustionSink: nothing proved the sink's log.warn actually invokes either
method (Cell A), and nothing proved it picks the right one for a profile
with vs without a configured model: (Cell B).

Extract the inline lambda into a new static Fleetd.exhaustionSink(...)
factory (same refactor-for-testability class the lead approved in round 2
for the two static warning methods), behaviourally unchanged from the
lambda it replaces. FleetdExhaustionSinkWarningTest drives this factory's
return value directly and asserts on the real text a ListAppender attached
to Fleetd's own logger captures, covering both cells from one mechanism:

- Cell A (ternary result replaced by a literal string): confirmed red,
  1594 run / 1 failure, restored, confirmed green (1594/0).
- Cell B (ternary's two branches swapped): confirmed red, 1594 run /
  2 failures (both test methods independently caught it), restored,
  confirmed green (1594/0).

.gitea/workflows/ci.yml

@@ -87,12 +87,18 @@ jobs:
           apt-get update && apt-get install -y --no-install-recommends maven
           mvn -version
 
-      # The `contract` profile clears the default-excludes group, so the @Tag("contract") AMQP test
-      # runs against the RabbitMQ service container (AMQP_URI). Pinned to the one contract test to
-      # avoid re-running the unit suite already covered by the `build` job.
+      # The `contract` profile clears the default-excludes group, so `-Dgroups=contract` runs every
+      # @Tag("contract") test and nothing from the unit suite the `build` job already covered — a
+      # tag selects the whole group, so a test added to it later runs here automatically. A prior
+      # version of this step pinned `-Dtest=AmqpReplyInboxContractTest` by class name instead: that
+      # silently excluded every other contract test (including the herdr ones) from CI, and nobody
+      # noticed until the herdr protocol drifted out from under a test that never ran here
+      # (fleetd #449). If this runner has no herdr socket, the herdr-backed tests in the group
+      # skip on their own `assumeTrue` and only the broker-backed ones actually run — check the
+      # step output rather than assuming which.
       - name: Contract tests
         working-directory: fleetd
-        run: mvn -B -Pcontract test -Dtest=AmqpReplyInboxContractTest
+        run: mvn -B -Pcontract test -Dgroups=contract
 
       - name: Failing test output
         if: failure()

fleetd/fleetd.example.yaml

@@ -207,8 +207,12 @@ herdrSocket: ~/.config/herdr/herdr.sock
 #                      omit and this profile's completion fallback behaves exactly as before.
 #                      Every backend words its refusal differently, so this is config, never a
 #                      vendor string baked into fleetd itself.
-#                      DEFERRED: compiled once into a startup pattern map — editing it needs a
-#                      daemon restart, same as this profile's model/baseUrl/argv.
+#                      HOT (fleetd #446): read live, cached by profile name, at every
+#                      completion-fallback check AND by fleet_profiles' exhaustionDetectionArmed —
+#                      editing it and reloading arms or disarms usage-limit detection for this
+#                      profile with no daemon restart. (Before fleetd #446 this was DEFERRED,
+#                      compiled once into a startup pattern map like model/baseUrl/argv still are —
+#                      see errorPattern below, which is still deferred that way on purpose.)
 #   credentialId     → CB-578 stage B: the credential this profile quarantines WITH when a
 #                      BACKEND_EXHAUSTED classification fires. Two profiles that set the SAME
 #                      credentialId share one quarantine — the case this exists for is two models
@@ -227,8 +231,9 @@ herdrSocket: ~/.config/herdr/herdr.sock
 #                      happens, just without a profile-specific match; every backend words its
 #                      failure differently, so a hardcoded sentence would only ever match one
 #                      of them.
-#                      DEFERRED: compiled once into a startup pattern map, same as exhaustedPattern
-#                      — editing it needs a daemon restart.
+#                      DEFERRED: compiled once into a startup pattern map — editing it needs a
+#                      daemon restart. Unlike exhaustedPattern above (made hot by fleetd #446),
+#                      errorPattern was scoped out of that ticket on purpose and stays deferred.
 #                      # errorPattern: "503 Service Unavailable"   # opt-in: classify a backend outage
 #
 #                      What happens once a match fires (BackendOutagePolicy, credentialId-keyed,
@@ -468,9 +473,10 @@ placement: weighted
 # when the reload happens — not about how important the key is:
 #   HOT       → takes effect on the next spawn: the whole `fleet:` block (every role pool,
 #               `charters`, and `tabLabel`), `placement:`, and an existing profile's weight / maxLoad
-#               / credentialId. Those are hot because the placement policy (and, for credentialId,
-#               the CB-578 stage B quarantine check) reads them through a supplier — being config is
-#               not by itself enough to make a key hot.
+#               / credentialId / exhaustedPattern. Those are hot because the placement policy (and,
+#               for credentialId, the CB-578 stage B quarantine check; for exhaustedPattern, fleetd
+#               #446's LiveExhaustedPatterns) reads them through a supplier — being config is not by
+#               itself enough to make a key hot.
 #               EXCEPT `fleet.leaders`: Fleetd.main reads it once at startup to build the lead tab
 #               scanner and launcher, and neither is rebuilt on reload. A changed/added/removed
 #               `fleet.leaders` entry is silently accepted — the reload reports "config reloaded"
@@ -482,10 +488,10 @@ placement: weighted
 #               stage B — baked once into the quarantine tracker built at startup), ADDING or
 #               REMOVING a profile (a new backend needs its own launcher, and launchers are built
 #               once), AND an existing profile's launch settings — model, baseUrl, argv, env,
-#               configDir, mcpUrl, tabLabel, exhaustedPattern, errorPattern (fleetd #201 / #227 —
-#               compiled once into a startup pattern map the same way exhaustedPattern is). The
-#               launcher takes a copy of `profiles:` at startup and resolves every spawn out of
-#               that copy, so those never
+#               configDir, mcpUrl, tabLabel, errorPattern (fleetd #201 / #227 — compiled once into
+#               a startup pattern map; exhaustedPattern used to be compiled the same way until
+#               fleetd #446 made it hot — see above). The launcher takes a copy of `profiles:` at
+#               startup and resolves every spawn out of that copy, so those never
 #               reach a launch until you restart. The reload logs them by name rather than
 #               pretending they applied.
 #   COLD      → cannot change at all: `bind:`, `herdrSocket:`, `broker:` and `auth:`. The socket is

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

@@ -18,6 +18,7 @@ import dev.ltms.fleet.inject.BackendErrorSink;
 import dev.ltms.fleet.inject.CompletionResolver;
 import dev.ltms.fleet.inject.ExhaustedPatternLookup;
 import dev.ltms.fleet.inject.ExhaustionSink;
+import dev.ltms.fleet.inject.LiveExhaustedPatterns;
 import dev.ltms.fleet.inject.Injector;
 import dev.ltms.fleet.inject.StatusPoller;
 import dev.ltms.fleet.inject.TurnListener;
@@ -65,11 +66,13 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -78,6 +81,7 @@ import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 /**
  * {@code fleetd} entry point. Wires the real herdr socket client to the REST app and
@@ -378,28 +382,34 @@ public final class Fleetd {
         Rendezvous rendezvous = new Rendezvous();
         // CB-578 stage A: classify a completion-fallback scrape that matches a profile's configured
         // usage-limit refusal as BACKEND_EXHAUSTED rather than handing it back as a real answer.
-        // Compiled once at startup, keyed by profile name; a profile with no exhaustedPattern is
-        // simply absent here, so its workers keep today's completion-fallback behaviour unchanged.
-        Map<String, Pattern> exhaustedPatternsByProfile = new LinkedHashMap<>();
-        cfg.profiles().forEach((name, profile) -> {
-            if (profile.hasExhaustedPattern()) {
-                exhaustedPatternsByProfile.put(name, Pattern.compile(profile.exhaustedPattern()));
-            }
-        });
+        // fleetd #446: read live off `config` per lookup, cached by profile name — see
+        // LiveExhaustedPatterns's class doc for why this replaces the old compiled-once-at-startup
+        // map. A profile with no exhaustedPattern simply returns null here, so its workers keep
+        // today's completion-fallback behaviour unchanged.
+        LiveExhaustedPatterns liveExhaustedPatterns = new LiveExhaustedPatterns(() -> config.get().profiles());
         ExhaustedPatternLookup exhaustedPatterns = target -> sessions.roster().stream()
                 .filter(session -> target.equals(session.terminalId()))
                 .findFirst()
-                .map(session -> exhaustedPatternsByProfile.get(session.profile()))
+                .map(session -> liveExhaustedPatterns.patternFor(session.profile()))
                 .orElse(null);
+        // The startup coverage line still reports the boot-time snapshot only — it is printed once,
+        // here, and a reload no longer needs to change what it said; exhaustionDetectionArmed (via
+        // liveExhaustedPatterns.armed, wired into quarantineSource below) is what stays live.
+        Set<String> exhaustedConfiguredAtStartup = cfg.profiles().entrySet().stream()
+                .filter(e -> e.getValue().hasExhaustedPattern())
+                .map(Map.Entry::getKey)
+                .collect(Collectors.toCollection(LinkedHashSet::new));
         log.info("backend-exhausted classification (CB-578 stage A): {}",
-                exhaustedPatternCoverageLine(cfg.profiles().keySet(), exhaustedPatternsByProfile.keySet()));
+                exhaustedPatternCoverageLine(cfg.profiles().keySet(), exhaustedConfiguredAtStartup));
         // fleetd #201 Unit 5: classify a completion-fallback scrape that matches a profile's
         // configured backend-error refusal (a credential outage, a provider 5xx) as a backend error
-        // rather than handing it back as a real answer. Compiled once at startup, keyed by profile
-        // name, mirroring exhaustedPatternsByProfile above — a profile with no configured
-        // errorPattern is simply absent here, so CompletionResolver falls back to its built-in
-        // narrow {@code (?i)\bAPI Error\s*:} compatibility pattern for that profile's targets
-        // (BackendErrorPatternLookup#legacy's contract — see backendErrorPatterns below).
+        // rather than handing it back as a real answer. Still compiled once at startup, keyed by
+        // profile name — unlike exhaustedPattern above (fleetd #446), errorPattern was left DEFERRED
+        // on purpose: the ticket that made exhaustedPattern hot scoped errorPattern/cooling-off out
+        // explicitly. A profile with no configured errorPattern is simply absent here, so
+        // CompletionResolver falls back to its built-in narrow {@code (?i)\bAPI Error\s*:}
+        // compatibility pattern for that profile's targets (BackendErrorPatternLookup#legacy's
+        // contract — see backendErrorPatterns below).
         Map<String, Pattern> errorPatternsByProfile = new LinkedHashMap<>();
         cfg.profiles().forEach((name, profile) -> {
             if (profile.hasErrorPattern()) {
@@ -417,46 +427,20 @@ public final class Fleetd {
         // CREDENTIAL — not the profile name — so a profile sharing that credential (e.g. two models
         // on one OpenAI account) is refused too, not just the one that happened to report it. Reads
         // the profile config live off `config`, so a credentialId edit is hot: no restart needed.
-        //
-        // fleetd #175: this sink is now also the quarantine target for OpenCodeLauncher's
-        // model-mismatch check — it needs nothing profile-specific from the caller beyond `target`
-        // (a herdr terminal id) and `reason`, so reusing it here is exactly "the existing
-        // ExhaustionSink path", not a new mechanism.
-        //
-        // fleetd #234: that check fires from SessionAwareHandle.agentSessionId(), which runs during
-        // SessionManager.acquire() BEFORE this session is registered in sessions.roster() — so the
-        // roster-only lookup below used to find nothing, .ifPresent silently no-op'd, and the
-        // ERROR the check had just logged ("quarantining this profile's credential") was a lie:
-        // nothing was quarantined, and nothing said so. Two changes: (1) OpenCodeLauncher now
-        // passes its OWN profile name via ExhaustionSink's 3-arg overload — it already has the
-        // FleetConfig.Profile in hand and does not need the roster at all — used here as a
-        // fallback whenever the roster lookup misses; (2) if a profile still cannot be resolved
-        // (neither the roster nor the hint names a configured one), this logs loudly at ERROR
-        // instead of silently doing nothing — a control that cannot act must say so.
-        // fleetd #234, round 4: the 3-arg overload is now ExhaustionSink's single abstract method,
-        // so this is safely a lambda — there is no separate 2-arg overload left for it to bind to
-        // instead and silently drop profileHint (that was rounds 1-3's whole hazard).
-        ExhaustionSink exhaustionSink = (target, reason, profileHint) -> {
-            String profileName = sessions.roster().stream()
-                    .filter(session -> target.equals(session.terminalId()))
-                    .findFirst()
-                    .map(MemberSession::profile)
-                    .orElse(profileHint);
-            FleetConfig.Profile profile = profileName == null ? null : config.get().profiles().get(profileName);
-            if (profile == null) {
-                log.error("quarantine requested for target '{}' ({}) but no profile could be "
-                                + "resolved — the target is not (yet) in the roster, and {} — "
-                                + "credential NOT quarantined (fleetd #234)",
-                        target, reason,
-                        profileHint == null ? "no profile hint was given"
-                                : "the hinted profile '" + profileHint + "' is not configured");
-                return;
-            }
-            String credentialId = profile.effectiveCredentialId();
-            quarantine.quarantine(credentialId);
-            log.warn("credential '{}' quarantined for {}s (profile '{}'): {}", credentialId,
-                    cfg.quarantineCooldownSeconds(), profile.profile(), reason);
-        };
+        // fleetd #446 criterion 3: the backend text that triggered the most recent quarantine of
+        // each credential, so fleet_profiles can report WHY a limit was hit, not only that it was.
+        // Keyed by credential id — the same key BackendQuarantine's own remainingSeconds uses —
+        // and written at the one call site that actually quarantines (inside exhaustionSink below),
+        // so a reason can never be reported for a quarantine that never happened. Bounded the same
+        // way BackendQuarantine's own internal map is documented to be: by the number of distinct
+        // credentials ever exhausted, not by how often the config is edited.
+        Map<String, String> quarantineReasonByCredential = new ConcurrentHashMap<>();
+        // fleetd #446 follow-up (round 3): extracted into exhaustionSink(...) below — see that
+        // method's javadoc for the full fleetd #175/#234/#446 history this used to carry inline —
+        // so a dedicated test can drive the exact ExhaustionSink main() builds, not a hand-rebuilt
+        // copy of its shape.
+        ExhaustionSink exhaustionSink = exhaustionSink(sessions, config, quarantine,
+                quarantineReasonByCredential, cfg);
         // fleetd #175: point the forwarding sink handed to OpenCodeLauncher above at the real one,
         // now that `sessions` exists to resolve target -> session -> profile.
         exhaustionSinkRef.set(exhaustionSink);
@@ -673,7 +657,7 @@ public final class Fleetd {
         // / 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 = quarantineSource(config, quarantine,
-                exhaustedPatternsByProfile);
+                liveExhaustedPatterns, quarantineReasonByCredential);
         FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(profile -> {
             var configured = config.get().profiles().get(profile);
             return configured == null ? null : configured.effectiveCredentialId();
@@ -805,16 +789,99 @@ public final class Fleetd {
     }
 
     /**
-     * fleetd #404: production source for quarantine reporting. Credential IDs are hot, but
-     * exhausted patterns are compiled once at startup for {@link CompletionResolver}, so the armed
-     * field must use that same compiled map until restart.
+     * fleetd #446 follow-up (round 3): the quarantine {@link ExhaustionSink} main() actually wires
+     * — extracted out of {@code main} for the same reason {@link #quarantineSource} below and
+     * {@link #usageLimitFixWarning}/{@link #usageLimitFixWarningNoModel} were. Round 2 pinned the
+     * WARNING text by having {@code FleetdUsageLimitFixWarningTest} call those two methods
+     * directly, but a mutation battery proved that left two gaps: nothing proved this sink's
+     * {@code log.warn} call actually uses either method (replacing the whole ternary with a
+     * literal string), and nothing proved it picks the right one for a profile WITH a configured
+     * {@code model:} versus one WITHOUT (swapping the ternary's two branches) — both mutations
+     * left round 2's full 1592-test suite green. {@code FleetdExhaustionSinkWarningTest} drives
+     * THIS factory's return value directly and asserts on the real text a {@code ListAppender}
+     * attached to this class's own logger captures — the only way to prove the caller, not just
+     * the two callees in isolation.
+     *
+     * <p>Behaviourally unchanged from the inline lambda this replaces:
+     * <ul>
+     *   <li>fleetd #175: also the quarantine target for {@code OpenCodeLauncher}'s model-mismatch
+     *       check — it needs nothing profile-specific beyond {@code target}/{@code reason}, so
+     *       reusing this sink is the existing {@code ExhaustionSink} path, not a new mechanism;</li>
+     *   <li>fleetd #234 (round 4): resolves {@code target} to a profile via the live session
+     *       roster first, falling back to {@code profileHint} — {@code
+     *       SessionAwareHandle.agentSessionId()} fires this sink during {@code
+     *       SessionManager.acquire()}, <em>before</em> that session is registered, so the
+     *       roster-only lookup alone used to silently miss it. A profile resolved neither way
+     *       logs loudly at ERROR instead of doing nothing;</li>
+     *   <li>fleetd #446 criterion 3: writes {@code quarantineReasonByCredential} at the one call
+     *       site that actually quarantines, keyed by credential id — see the field's declaration
+     *       in {@code main} for the bound on its size.</li>
+     * </ul>
+     *
+     * <p>{@code cfg} is the startup {@link FleetConfig} snapshot, read only for {@code
+     * quarantineCooldownSeconds()} in the log text — a Cold key (see {@code ConfigRef}'s class
+     * doc), so reading it off the snapshot rather than {@code config.get()} makes no observable
+     * difference and matches what the inline version already did.
+     */
+    static ExhaustionSink exhaustionSink(SessionManager sessions, ConfigRef config, BackendQuarantine quarantine,
+            Map<String, String> quarantineReasonByCredential, FleetConfig cfg) {
+        return (target, reason, profileHint) -> {
+            String profileName = sessions.roster().stream()
+                    .filter(session -> target.equals(session.terminalId()))
+                    .findFirst()
+                    .map(MemberSession::profile)
+                    .orElse(profileHint);
+            FleetConfig.Profile profile = profileName == null ? null : config.get().profiles().get(profileName);
+            if (profile == null) {
+                log.error("quarantine requested for target '{}' ({}) but no profile could be "
+                                + "resolved — the target is not (yet) in the roster, and {} — "
+                                + "credential NOT quarantined (fleetd #234)",
+                        target, reason,
+                        profileHint == null ? "no profile hint was given"
+                                : "the hinted profile '" + profileHint + "' is not configured");
+                return;
+            }
+            String credentialId = profile.effectiveCredentialId();
+            quarantine.quarantine(credentialId);
+            quarantineReasonByCredential.put(credentialId, reason);
+            log.warn("credential '{}' quarantined for {}s (profile '{}'): {}", credentialId,
+                    cfg.quarantineCooldownSeconds(), profile.profile(), reason);
+            // fleetd #446 criterion 2: name the fix, not just the fact — an operator reading this
+            // should not have to work out which of several configured models to touch.
+            String model = profile.model();
+            log.warn(model != null && !model.isBlank()
+                    ? usageLimitFixWarning(profile.profile(), model)
+                    : usageLimitFixWarningNoModel(profile.profile(), cfg.quarantineCooldownSeconds()));
+        };
+    }
+
+    /**
+     * fleetd #404, superseded by fleetd #446: production source for quarantine reporting.
+     * Credential IDs were already hot; {@code exhaustedPattern} used to be compiled once at
+     * startup for {@link CompletionResolver}, so the armed field had to read that same frozen map
+     * until restart — the exact asymmetry #446 exists to close. Both {@code exhaustedPatternArmed}
+     * here and {@link CompletionResolver}'s own classification now read the ONE live {@link
+     * LiveExhaustedPatterns} instance ({@code exhaustedPatterns.armed}/{@code .patternFor}), so a
+     * reload that arms or disarms a profile's detection is visible to both at once — never two
+     * independently-updated copies that could disagree, the fleetd #404 rule this method used to
+     * violate on purpose and now upholds.
+     *
+     * <p>{@code modelFor} and {@code reasonFor} back fleetd #446 criterion 3 — {@code
+     * fleet_profiles}'s quarantined row naming which model a quarantined profile runs
+     * ({@code modelFor}, read live off {@code config} the same way {@code credentialIdFor} already
+     * is) and the backend text that triggered the most recent quarantine of that credential
+     * ({@code reasonFor}, backed by {@code reasonByCredential} — see its call site in {@code main}
+     * for where that map is written, at the one place a credential is actually quarantined).
      */
     static FleetMcp.QuarantineSource quarantineSource(ConfigRef config, BackendQuarantine quarantine,
-            Map<String, Pattern> startupExhaustedPatterns) {
+            LiveExhaustedPatterns exhaustedPatterns, Map<String, String> reasonByCredential) {
         return new FleetMcp.QuarantineSource(profile -> {
             var configured = config.get().profiles().get(profile);
             return configured == null ? null : configured.effectiveCredentialId();
-        }, quarantine, profile -> startupExhaustedPatterns.containsKey(profile));
+        }, quarantine, exhaustedPatterns::armed, profile -> {
+            var configured = config.get().profiles().get(profile);
+            return configured == null ? null : configured.model();
+        }, reasonByCredential::get);
     }
 
     /**
@@ -833,6 +900,40 @@ public final class Fleetd {
      * with 0 errors and left all 1506 existing tests green before {@code
      * FleetdPatternCoverageLineTest} was added to catch exactly that swap.
      */
+    /**
+     * fleetd #446 follow-up: the criterion-2 WARNING text — "name the fix, not just the fact" —
+     * for a profile whose {@code model:} is configured. Extracted out of the {@code
+     * exhaustionSink} lambda the same way {@link #exhaustedPatternCoverageLine} was extracted out
+     * of {@code main}: the inline version compiled, ran, and read correctly, but nothing pinned
+     * its text, so a later edit could silently stop naming the fix and every test would stay
+     * green (measured: renaming the leading {@code "usage-limit fix:"} tag left all 1586
+     * pre-follow-up tests passing). {@link FleetdUsageLimitFixWarningTest} asserts on the return
+     * value of this method directly, which is exactly what the caller logs — {@code log.warn} is
+     * called with this method's result as a single, already-formatted argument, so the string a
+     * test sees here is byte-identical to what {@code fleetd.out} receives.
+     *
+     * <p>Deliberately asserts on three substrings, not the whole sentence (the profile name, the
+     * model name, and the literal {@code enabled: false} / {@code models.allow} action) — see that
+     * test's class doc for why a whole-sentence pin is the wrong granularity here.
+     */
+    static String usageLimitFixWarning(String profile, String model) {
+        return "usage-limit fix: profile '" + profile + "' runs model '" + model + "' — set "
+                + "`enabled: false` on that model's entry under models.allow in fleetd.yaml to "
+                + "stop new spawns landing on it (models: is hot, no restart needed); remove the "
+                + "line again once the subscription window resets";
+    }
+
+    /**
+     * fleetd #446 follow-up: the {@link #usageLimitFixWarning} counterpart for a profile with no
+     * {@code model:} configured — {@code models.allow} gates by model name, so there is nothing
+     * for an operator to flip, and this says so instead of naming a fix that does not exist.
+     */
+    static String usageLimitFixWarningNoModel(String profile, Integer quarantineCooldownSeconds) {
+        return "usage-limit fix: profile '" + profile + "' has no model: configured, so "
+                + "models.allow cannot gate it by name — the " + quarantineCooldownSeconds
+                + "s quarantine above is the only thing keeping new spawns off it for now";
+    }
+
     static String exhaustedPatternCoverageLine(Set<String> allProfiles, Set<String> configuredProfiles) {
         return CompletionResolver.coverage("exhaustedPattern", CompletionResolver.UnsetMeaning.OFF,
                 allProfiles, configuredProfiles);

fleetd/src/main/java/dev/ltms/fleet/config/ConfigRef.java

@@ -51,7 +51,17 @@ import java.util.function.Supplier;
  *       filter) and by {@code fleet_profiles}/{@code GET /profiles} (via
  *       {@code PeerLauncher.disabledModels()}). Nothing about {@code models:} is baked into an
  *       object built at startup, so — unlike the deferred keys below — there is no frozen half left
- *       to report; it moved here from deferred rather than joining split.</li>
+ *       to report; it moved here from deferred rather than joining split. An existing profile's
+ *       {@code exhaustedPattern} (fleetd #446) joined this class the same way: it used to be
+ *       compiled once into {@code Fleetd.main}'s startup pattern map (see the Deferred bullet's old
+ *       wording, and {@code LiveExhaustedPatterns}'s class doc for the history), and is now read
+ *       live, cached by profile name, by both {@code CompletionResolver}'s classification (via
+ *       {@code LiveExhaustedPatterns.patternFor}) and {@code fleet_profiles}'s {@code
+ *       exhaustionDetectionArmed} (via {@code LiveExhaustedPatterns.armed}) — the one live object
+ *       both read, so a reload that arms or disarms a profile's usage-limit detection takes effect
+ *       on the next check with no restart. {@code errorPattern}, {@code exhaustedPattern}'s sibling
+ *       key for backend-error (not usage-limit) classification, was deliberately left OUT of this
+ *       fleetd #446 change and stays deferred below — the ticket scoped it out explicitly.</li>
  *   <li><strong>Deferred</strong> — accepted into the new snapshot, but the wiring built at startup
  *       keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:},
  *       {@code idleSleepGuard:} ({@code Fleetd.java} reads it once, at startup, to decide whether
@@ -81,15 +91,17 @@ import java.util.function.Supplier;
  *       (if any) simply keeps its old settings), adding or removing a profile (a new backend needs its own launcher,
  *       which is constructed once), <em>and an existing profile's launch settings</em> —
  *       {@code model}, {@code baseUrl}, {@code argv}, {@code env}, {@code mcpUrl},
- *       {@code exhaustedPattern} (CB-578 stage A — compiled once into {@code Fleetd.main}'s
- *       pattern map at startup), {@code errorPattern} (fleetd #201 Unit 5 — compiled once into
- *       {@code Fleetd.main}'s backend-error pattern map at startup, the same way),
+ *       {@code errorPattern} (fleetd #201 Unit 5 — compiled once into
+ *       {@code Fleetd.main}'s backend-error pattern map at startup; deliberately NOT made hot
+ *       alongside {@code exhaustedPattern} by fleetd #446 — that ticket scoped {@code errorPattern}
+ *       and cooling-off out explicitly),
  *       {@code ideProjectDir} / {@code ideOpenCommand} / {@code autoCompactWindow} (fleetd #323
  *       instance 1 — all three are read at spawn off the same frozen profile map and were missing
  *       from {@link #sameLaunchSettings}), and the rest of {@link #sameLaunchSettings}.
- *       {@code credentialId} (CB-578 stage B) is NOT on
- *       this list — it is read live off the config supplier at every quarantine check and
- *       exhaustion event, exactly like {@code weight} / {@code maxLoad}, so it is hot instead.
+ *       {@code credentialId} (CB-578 stage B) and {@code exhaustedPattern} (fleetd #446) are NOT on
+ *       this list — both are read live off the config supplier at every quarantine check and
+ *       exhaustion event, exactly like {@code weight} / {@code maxLoad}, so both are hot instead
+ *       (see the Hot bullet above for {@code exhaustedPattern}'s history).
  *       {@code HerdrPeerLauncher} takes {@code Map.copyOf(profiles)} at construction and resolves
  *       each spawn out of that copy, so those never reach a launch until the daemon restarts. A
  *       reload logs these rather than pretending they applied.</li>
@@ -583,12 +595,16 @@ public final class ConfigRef implements Supplier<FleetConfig> {
      * {@link #sameLaunchSettings} because they are read <em>live</em>, not baked in at spawn — see
      * the class doc's <em>Hot</em> bullet. {@code weight} and {@code maxLoad} are read live by the
      * placement policy on every spawn; {@code credentialId} is read live by
-     * {@code CompositePeerLauncher} and the CB-578 stage B exhaustion sink. Nothing else is
+     * {@code CompositePeerLauncher} and the CB-578 stage B exhaustion sink; {@code exhaustedPattern}
+     * (fleetd #446) is read live, cached by profile name, by {@code LiveExhaustedPatterns} — see
+     * that class's doc and the class doc's <em>Hot</em> bullet for the history (it used to be
+     * compared here, deferred, like its sibling {@code errorPattern} still is). Nothing else is
      * excluded — see {@code sameLaunchSettingsComparesEveryProfileComponentOrExcludesIt} in
      * {@code ConfigRefProfileCoverageTest}, which enumerates every {@code Profile} record component
      * by reflection and fails the build if one is neither compared below nor named here.
      */
-    static final Set<String> LAUNCH_SETTINGS_EXCLUDED = Set.of("weight", "maxLoad", "credentialId");
+    static final Set<String> LAUNCH_SETTINGS_EXCLUDED =
+            Set.of("weight", "maxLoad", "credentialId", "exhaustedPattern");
 
     /**
      * Whether two versions of a profile would launch a peer identically.
@@ -625,13 +641,15 @@ public final class ConfigRef implements Supplier<FleetConfig> {
                 && Objects.equals(a.kind(), b.kind())
                 && Objects.equals(a.env(), b.env())
                 && Objects.equals(a.subscription(), b.subscription())
-                // CB-578 stage B: exhaustedPattern is compiled once into Fleetd.main's pattern map
-                // at startup (see ExhaustedPatternLookup wiring) — a reload never re-reads it, so a
-                // changed pattern must be reported as deferred, exactly like model/baseUrl/argv.
-                && Objects.equals(a.exhaustedPattern(), b.exhaustedPattern())
+                // fleetd #446: exhaustedPattern moved to LAUNCH_SETTINGS_EXCLUDED — it is now read
+                // live, cached by profile name, through LiveExhaustedPatterns (see that class's doc
+                // and ConfigRef's class doc Hot bullet), so it must NOT be compared here any more: a
+                // reload that changes only exhaustedPattern must report "config reloaded", not
+                // "these changes need a restart".
                 // fleetd #201 Unit 5: errorPattern is compiled once into Fleetd.main's backend-error
-                // pattern map at startup (see BackendErrorPatternLookup wiring), the same way
-                // exhaustedPattern is — a reload never re-reads it either.
+                // pattern map at startup (see BackendErrorPatternLookup wiring) — unlike its sibling
+                // exhaustedPattern above (fleetd #446), a reload still never re-reads it; fleetd #446
+                // scoped errorPattern out on purpose (see the class doc's Hot bullet).
                 && Objects.equals(a.errorPattern(), b.errorPattern())
                 // fleetd #323 instance 1: ideProjectDir and ideOpenCommand are read at spawn off the
                 // same frozen profile map as ideMcpUrl above (ClaudeCodeLauncher.java:267/269,

fleetd/src/main/java/dev/ltms/fleet/config/FleetConfig.java

@@ -424,6 +424,12 @@ public record FleetConfig(
      *                       {@code null}/blank ⇒ the classification never fires for this profile and
      *                       today's completion-fallback behaviour is unchanged. Every backend words
      *                       its refusal differently, so this is config, never a vendor string in code.
+     *                       Read live off the current config through {@code LiveExhaustedPatterns},
+     *                       cached by profile name (fleetd #446), so it is HOT: an operator can arm
+     *                       or disarm this profile's usage-limit detection by editing this key and
+     *                       reloading, with no restart. Before fleetd #446 this was compiled once at
+     *                       daemon startup, the same deferred shape its sibling {@code errorPattern}
+     *                       (below) still has.
      * @param credentialId   the shared account this profile authenticates as (CB-578 stage B). Two
      *                       or more profiles setting the <em>same</em> non-blank value are quarantined
      *                       together by one {@code exhaustedPattern} classification on any one of

fleetd/src/main/java/dev/ltms/fleet/herdr/HerdrClient.java

@@ -3,7 +3,7 @@ package dev.ltms.fleet.herdr;
 import com.fasterxml.jackson.databind.JsonNode;
 
 /**
- * Client face onto the herdr daemon (protocol 14, herdr 0.7.0).
+ * Client face onto the herdr daemon (protocol 19, herdr 0.8.0).
  *
  * <p>This is the ONLY thing in {@code fleetd} that speaks to herdr. Every method
  * maps to a herdr JSON-RPC call over its Unix domain socket. Requests are

fleetd/src/main/java/dev/ltms/fleet/herdr/HerdrCodec.java

@@ -8,7 +8,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.nio.charset.StandardCharsets;
 
 /**
- * Wire codec for herdr's newline-delimited JSON-RPC (protocol 14).
+ * Wire codec for herdr's newline-delimited JSON-RPC (protocol 19).
  *
  * <p>Split out from the socket so the framing rules — the ones that actually bit us
  * during the spike (id MUST be a string; response carries {@code result} or

fleetd/src/main/java/dev/ltms/fleet/inject/LiveExhaustedPatterns.java

@@ -0,0 +1,91 @@
+package dev.ltms.fleet.inject;
+
+import dev.ltms.fleet.config.FleetConfig;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+import java.util.regex.Pattern;
+
+/**
+ * fleetd #446: the single live source for "is {@code exhaustedPattern} configured for this
+ * profile, and what does it compile to". Read fresh off the config supplier on every call — the
+ * same reason {@code CompositePeerLauncher#models0} is a live supplier read rather than a value
+ * captured at construction (see that class's doc) — so an operator can arm or disarm usage-limit
+ * detection for a profile by editing {@code exhaustedPattern} and reloading, with no restart.
+ *
+ * <p>Before this class, {@code exhaustedPattern} was compiled once into a {@code Map<String,
+ * Pattern>} built inside {@code Fleetd.main} at startup ({@code ConfigRef}'s class doc used to
+ * list it under <em>Deferred</em>, CB-578 stage A) — the asymmetry fleetd #446 exists to close:
+ * an operator could turn a model off at runtime ({@code models.allow}'s {@code enabled: false},
+ * hot since fleetd #422) but could not arm the detector that would tell them to, without a
+ * restart. That was backwards for a feature whose whole point is to react while the fleet runs.
+ *
+ * <p>{@link #patternFor} is what {@link CompletionResolver} enforces on (via the {@link
+ * ExhaustedPatternLookup} production wiring in {@code Fleetd.main}); {@link #armed} is what
+ * {@code fleet_profiles}/{@code GET /profiles} report as {@code exhaustionDetectionArmed}. Both
+ * read this ONE object, so the report can never disagree with the behaviour — the same rule
+ * {@code CompositePeerLauncher#modelGateState()}'s javadoc states for the model gate's own
+ * armed/off pair (fleetd #404): "armed" and "which models are off" must come from one read of the
+ * same accessor the gate enforces on.
+ *
+ * <h2>Cache eviction</h2>
+ * Cached by profile NAME, not by pattern text. A {@code Map<String, Pattern>} keyed by the
+ * pattern STRING would grow by one entry per distinct regex ever typed for any profile across the
+ * daemon's uptime — unbounded in practice, because tuning a regex to match a backend's exact
+ * wording is exactly the kind of edit an operator makes several times while getting it right, and
+ * every edit-and-reload cycle would leave the previous attempt's compiled {@link Pattern} behind
+ * forever. Keyed by profile name instead, this cache holds at most one entry per profile name that
+ * has ever been looked up — and that key space is bounded by the (small, human-authored) set of
+ * configured profiles, which changes only on a restart: adding or removing a profile is itself a
+ * deferred key (a new backend needs its own launcher, built once — see {@code ConfigRef}'s class
+ * doc), so profile names do not churn the way pattern text does. Re-editing an EXISTING profile's
+ * {@code exhaustedPattern} — the case this class exists to make hot — simply overwrites that
+ * profile's one cache entry; it never adds a new one.
+ */
+public final class LiveExhaustedPatterns {
+
+    /** A compiled pattern paired with the source string it was compiled from, for change detection. */
+    private record Cached(String source, Pattern compiled) {
+    }
+
+    private final Supplier<Map<String, FleetConfig.Profile>> profiles;
+    private final ConcurrentHashMap<String, Cached> cache = new ConcurrentHashMap<>();
+
+    public LiveExhaustedPatterns(Supplier<Map<String, FleetConfig.Profile>> profiles) {
+        this.profiles = Objects.requireNonNull(profiles, "profiles");
+    }
+
+    /**
+     * The compiled {@code exhaustedPattern} currently configured for {@code profileName}, or
+     * {@code null} when that profile is unknown or has none configured. Recompiles only when the
+     * live pattern text differs from what is cached for this profile name; {@code
+     * FleetConfig#rejectMalformedProfilePatterns} already refuses a config (at load and at reload)
+     * whose {@code exhaustedPattern} does not compile, so this is not expected to throw in
+     * production — it is not defended against here for that reason, the same trust
+     * {@code CompositePeerLauncher#models0} places in config validation having already run.
+     */
+    public Pattern patternFor(String profileName) {
+        if (profileName == null) {
+            return null;
+        }
+        FleetConfig.Profile profile = profiles.get().get(profileName);
+        if (profile == null || !profile.hasExhaustedPattern()) {
+            return null;
+        }
+        String source = profile.exhaustedPattern();
+        Cached cached = cache.get(profileName);
+        if (cached != null && cached.source().equals(source)) {
+            return cached.compiled();
+        }
+        Cached fresh = new Cached(source, Pattern.compile(source));
+        cache.put(profileName, fresh);
+        return fresh.compiled();
+    }
+
+    /** Whether {@code profileName} currently has a usage-limit pattern configured (live). */
+    public boolean armed(String profileName) {
+        return patternFor(profileName) != null;
+    }
+}

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

@@ -121,17 +121,46 @@ public final class FleetMcp {
      * CB-578 stage B quarantine facts used by {@code fleet_profiles}: a profile → credential id
      * lookup, plus the shared {@link BackendQuarantine} to read remaining cooldowns off.
      *
-     * @param exhaustedPatternArmed fleetd #395: profile → whether that profile's {@code
-     *                      exhaustedPattern} is configured (see {@code
+     * @param exhaustedPatternArmed fleetd #395, made live by fleetd #446: profile → whether that
+     *                      profile's {@code exhaustedPattern} is configured right now (see {@code
      *                      FleetConfig.Profile#hasExhaustedPattern}), i.e. whether a backend refusal
      *                      on it can EVER be classified {@code BACKEND_EXHAUSTED} and quarantine its
      *                      credential. Bundled here, not a separate Source, because it answers the
      *                      exact question {@code fleet_profiles}'s quarantine facts already answer
      *                      for a QUARANTINED profile — "can this profile's usage limit ever be
-     *                      caught?" — just for every profile, not only one currently caught.
+     *                      caught?" — just for every profile, not only one currently caught. In
+     *                      production this reads {@link dev.ltms.fleet.inject.LiveExhaustedPatterns#armed} —
+     *                      the SAME live accessor {@link dev.ltms.fleet.inject.CompletionResolver}
+     *                      enforces on — so a reload that arms or disarms detection is reported
+     *                      correctly on the very next call, no restart (see that class's doc).
+     * @param modelFor      fleetd #446 criterion 3: profile → the {@code model:} it runs, or
+     *                      {@code null} when the profile names none. Read live off the current
+     *                      config, the same way {@code credentialIdFor} already is. Used to name,
+     *                      in a QUARANTINED profile's row, which of the operator's configured models
+     *                      the fix (\"set {@code enabled: false} on it under {@code models.allow}\")
+     *                      actually applies to.
+     * @param reasonFor     fleetd #446 criterion 3: credential id → the backend text that triggered
+     *                      its most recent quarantine, or {@code null} when none is known (e.g. an
+     *                      inert source, or a quarantine recorded before this field existed). Never
+     *                      consulted on its own to decide whether a credential is quarantined —
+     *                      callers gate on {@link BackendQuarantine#remainingSeconds} first, exactly
+     *                      like {@code credentialId} itself, so a stale reason left behind after a
+     *                      quarantine expires is never surfaced.
      */
     public record QuarantineSource(Function<String, String> credentialIdFor, BackendQuarantine quarantine,
-                                    Function<String, Boolean> exhaustedPatternArmed) {
+                                    Function<String, Boolean> exhaustedPatternArmed,
+                                    Function<String, String> modelFor,
+                                    Function<String, String> reasonFor) {
+        /**
+         * Backward-compatible 3-arg form, before fleetd #446 added {@code modelFor}/{@code
+         * reasonFor} — neither is ever reported. Keeps every pre-existing call site (production and
+         * test) compiling and behaving identically for the quarantine facts they actually asked for.
+         */
+        public QuarantineSource(Function<String, String> credentialIdFor, BackendQuarantine quarantine,
+                                 Function<String, Boolean> exhaustedPatternArmed) {
+            this(credentialIdFor, quarantine, exhaustedPatternArmed, _ -> null, _ -> null);
+        }
+
         /**
          * Backward-compatible 2-arg form, before fleetd #395 added {@code exhaustedPatternArmed} —
          * reports every profile unarmed. Keeps every pre-existing call site (production and test)
@@ -994,7 +1023,11 @@ public final class FleetMcp {
         if (isBlank(target) || isBlank(msgId)) {
             return error("target and msgId are required");
         }
-        messages.ackReply(target, msgId);
+        if (!messages.ackReply(target, msgId)) {
+            return error(msgId + " is not in " + target + "'s reply inbox (wrong id, wrong target, "
+                    + "or already acked). Held lead-to-lead (peer) mail cannot be acked this way — "
+                    + "read it with fleet_poll{coordId}.");
+        }
         return text("acknowledged " + msgId);
     }
 
@@ -1206,13 +1239,21 @@ public final class FleetMcp {
      * #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.
      *
-     * <p>fleetd #395: also reports {@code exhaustionDetectionArmed}, one boolean per configured
-     * profile — {@code true} when that profile's {@code exhaustedPattern} is set, {@code false}
-     * when it is not, so an operator can tell "this profile is healthy" from "nothing can ever
-     * quarantine this profile" without reading {@code fleetd.yaml}. Unlike {@code quarantined}/
-     * {@code coolingOff}, this map always names every profile: an unarmed profile never enters a
-     * transient state to be absent from, so silence here would read as "healthy" rather than "not
-     * being watched at all".
+     * <p>fleetd #395, made live by fleetd #446: also reports {@code exhaustionDetectionArmed}, one
+     * boolean per configured profile — {@code true} when that profile's {@code exhaustedPattern} is
+     * set RIGHT NOW, {@code false} when it is not, so an operator can tell "this profile is
+     * healthy" from "nothing can ever quarantine this profile" without reading {@code fleetd.yaml}.
+     * Unlike {@code quarantined}/{@code coolingOff}, this map always names every profile: an
+     * unarmed profile never enters a transient state to be absent from, so silence here would read
+     * as "healthy" rather than "not being watched at all". Hot since fleetd #446: a reload that
+     * arms or disarms a profile's {@code exhaustedPattern} changes this map's answer on the very
+     * next call, no restart — see {@link dev.ltms.fleet.inject.LiveExhaustedPatterns}'s class doc.
+     *
+     * <p>fleetd #446 criterion 3: each {@code quarantined} row also names {@code model} (the
+     * profile's configured {@code model:}, omitted when the profile names none) and {@code reason}
+     * (the backend text that triggered the most recent quarantine of that credential, omitted when
+     * none is known) — so a lead can see WHICH model to turn off and WHY, without reading the
+     * daemon log.
      */
     public static Map<String, Object> profilesView(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
         Map<String, Object> result = new LinkedHashMap<>();
@@ -1229,6 +1270,14 @@ public final class FleetMcp {
                     Map<String, Object> row = new LinkedHashMap<>();
                     row.put("credentialId", credentialId);
                     row.put("quarantinedForSeconds", remaining);
+                    String model = quarantine.modelFor().apply(profile);
+                    if (model != null && !model.isBlank()) {
+                        row.put("model", model);
+                    }
+                    String reason = quarantine.reasonFor().apply(credentialId);
+                    if (reason != null && !reason.isBlank()) {
+                        row.put("reason", reason);
+                    }
                     quarantined.put(profile, row);
                 });
             }
@@ -1766,7 +1815,10 @@ public final class FleetMcp {
                         + "has processed a reply and wants to confirm it, leaving other pending replies "
                         + "in the inbox for later drain.",
                 objectSchema(Map.of(
-                        "target", stringProp("Worker session id whose inbox to ack from"),
+                        "target", stringProp("Worker session id whose inbox to ack from. Must name a "
+                                + "reply actually queued for it — an id in no inbox, or a coord-id "
+                                + "(peer held mail, read with fleet_poll{coordId} instead), errors "
+                                + "rather than reporting a false success"),
                         "msgId", stringProp("The message id to acknowledge")),
                         List.of("target", "msgId")));
     }
@@ -1808,11 +1860,16 @@ public final class FleetMcp {
                 "List the configured worker profiles (backends) and which one fleet_spawn uses by "
                         + "default. A 'quarantined' map is present when a backend-exhausted refusal put "
                         + "a profile's credential on cooldown — fleet_spawn onto it is refused until "
-                        + "quarantinedForSeconds elapses; a profile sharing that credential is listed too. "
+                        + "quarantinedForSeconds elapses; a profile sharing that credential is listed too, "
+                        + "each row naming 'model' (the model that profile runs, when configured) and "
+                        + "'reason' (the backend text that triggered the quarantine, when known) — the fix "
+                        + "is usually `enabled: false` on that model under models.allow. "
                         + "'exhaustionDetectionArmed' reports, per profile, whether a usage-limit refusal "
                         + "on it can EVER be classified and quarantined (its exhaustedPattern is "
                         + "configured) — false means that profile's credential can never be quarantined "
-                        + "by this mechanism, however many usage-limit refusals it sees.",
+                        + "by this mechanism, however many usage-limit refusals it sees. Both "
+                        + "exhaustedPattern and models.allow's on/off state are hot: editing fleetd.yaml "
+                        + "and reloading arms/disarms detection or flips a model off with no restart.",
                 objectSchema(Map.of(), List.of()));
     }
 

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

@@ -14,6 +14,7 @@ import dev.ltms.fleet.placement.BackendOutagePolicy;
 import dev.ltms.fleet.placement.BackendQuarantine;
 import dev.ltms.fleet.placement.PlacementCandidate;
 import dev.ltms.fleet.placement.PlacementContext;
+import dev.ltms.fleet.placement.PlacementDecision;
 import dev.ltms.fleet.placement.PlacementException;
 import dev.ltms.fleet.placement.PlacementPolicies;
 import dev.ltms.fleet.placement.PlacementPolicy;
@@ -402,21 +403,10 @@ public final class CompositePeerLauncher implements PeerLauncher {
         // the whole profile list. An EXPLICIT profile (above) is left alone on purpose — it is the
         // operator overriding, and refusing it would break `fleet_spawn{profile:"opus"}`, which
         // carries no role and so would be judged against the dev pool it was never meant for.
-        List<PlacementCandidate> candidates = candidates(req.role());
-        String roleDefault = defaultProfileFor(req.role());
         Set<String> unreachable = new HashSet<>();
-        // CB-578 stage B: computed once up front — a quarantine's expiry cannot pass within one spawn
-        // call, so re-deriving it per retry would only cost work, never change the answer.
-        Set<String> quarantined = quarantinedProfiles(candidates);
-        // fleetd #201 Unit 5: a distinct set from quarantined — see PlacementContext.coolingOff.
-        Set<String> coolingOff = coolingOffProfiles(candidates);
-        // fleetd #422: read live per spawn, same as quarantined/coolingOff above — a config reload
-        // that flips a model's enabled state is visible to the very next unqualified spawn.
-        Set<String> modelOff = modelOffProfiles(candidates);
-        PlacementContext ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable,
-                quarantined, coolingOff, modelOff);
+        PlacementContext ctx = placementContextFor(req.role(), unreachable);
 
-        int maxAttempts = candidates.isEmpty() ? 1 : candidates.size();
+        int maxAttempts = ctx.candidates().isEmpty() ? 1 : ctx.candidates().size();
         for (int attempt = 0; attempt < maxAttempts; attempt++) {
             // Deliberately uncaught: when no candidate is left (all at cap, or all unreachable) the
             // policy already throws a clear message. Catching it to rethrow a generic
@@ -443,8 +433,7 @@ public final class CompositePeerLauncher implements PeerLauncher {
                         chosen.profile(), e.getMessage());
                 unreachable.add(chosen.profile());
                 // Update the context for the next selection so the policy excludes this profile.
-                ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable,
-                        quarantined, coolingOff, modelOff);
+                ctx = placementContextFor(req.role(), unreachable);
             }
         }
 
@@ -615,8 +604,23 @@ public final class CompositePeerLauncher implements PeerLauncher {
         return known.isEmpty() ? List.copyOf(configured.keySet()) : known;
     }
 
-    /** The profile an unqualified spawn for {@code role} falls back to under {@code fixed} placement. */
-    private String defaultProfileFor(MemberRole role) {
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Live: reads {@link #poolFor}, which reads {@link #profileConfigs} and {@link #fleet} fresh
+     * on every call, so a config reload is visible without a restart (fleetd #425) — unlike {@link
+     * #defaultProfile}, the field captured once at construction, which this falls back to only when
+     * {@link #poolFor} has nothing to offer at all (no profiles configured for this composite).
+     *
+     * <p>Exact only under the {@code fixed} placement policy — the one that reads this value
+     * ({@code FixedPlacementPolicy}, package-private, hence not linked) as its first, preferred
+     * candidate. {@code weighted}/{@code round-robin} placement can choose a different candidate
+     * from {@code role}'s pool even on the very first spawn; this method does not simulate that
+     * choice, matching what the {@code defaultProfile:}-derived reporting this replaces has always
+     * done.
+     */
+    @Override
+    public String defaultProfileFor(MemberRole role) {
         List<String> pool = poolFor(role);
         return pool.isEmpty() ? defaultProfile : pool.getFirst();
     }
@@ -633,6 +637,142 @@ public final class CompositePeerLauncher implements PeerLauncher {
         return out;
     }
 
+    /**
+     * Build the {@link PlacementContext} an unqualified spawn of {@code role} would be judged
+     * against right now — the single source both {@link #spawn} and {@link #place} read, so the two
+     * can never disagree about which conditions (quarantine, cool-off, model-off) apply to which
+     * candidate (fleetd #425 rework: round 1 duplicated this into a second, blind resolver —
+     * {@link #defaultProfileFor} — which is why it regressed; round 2 found that even a single
+     * shared resolver is not enough on its own if the CALLER re-resolves through an explicit
+     * profile afterwards — see {@link PlacementDecision}).
+     *
+     * @param unreachable the caller's mutable unreachable set; {@link #spawn} grows this across
+     *                    retries and rebuilds the context from it, {@link #place} passes a fresh
+     *                    empty one since it never retries
+     */
+    private PlacementContext placementContextFor(MemberRole role, Set<String> unreachable) {
+        List<PlacementCandidate> candidates = candidates(role);
+        String roleDefault = defaultProfileFor(role);
+        // CB-578 stage B: computed once up front — a quarantine's expiry cannot pass within one spawn
+        // call, so re-deriving it per retry would only cost work, never change the answer.
+        Set<String> quarantined = quarantinedProfiles(candidates);
+        // fleetd #201 Unit 5: a distinct set from quarantined — see PlacementContext.coolingOff.
+        Set<String> coolingOff = coolingOffProfiles(candidates);
+        // fleetd #422: read live per spawn, same as quarantined/coolingOff above — a config reload
+        // that flips a model's enabled state is visible to the very next unqualified spawn.
+        Set<String> modelOff = modelOffProfiles(candidates);
+        return new PlacementContext(roleDefault, candidates, liveCount, unreachable,
+                quarantined, coolingOff, modelOff);
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * <p>fleetd #425 rework, round 2: runs the exact same selection {@link #spawn} uses for a
+     * blank-profile request — {@link #placementContextFor} plus one {@link PlacementPolicy#select}
+     * — rather than {@link #defaultProfileFor}'s blind "pool's first entry", so a quarantined,
+     * cooling-off, or model-off pool-first candidate is routed around here exactly as it would be
+     * by a real spawn. Unlike {@link #spawn}, this never retries on {@link
+     * PeerUnreachableException}: there is no spawn attempt to fail, so "unreachable" never grows
+     * past the empty set it starts with, and a single {@link PlacementPolicy#select} call already
+     * reflects the live quarantine/cool-off/model-off state.
+     *
+     * <p>Deliberately does <em>not</em> apply {@link #enforceMaxLoad} (or any of the other three
+     * {@code enforce*} checks): those belong to {@link #spawn}'s EXPLICIT-profile branch, the
+     * operator-override path, and this method answers a different question — "where would an
+     * UNQUALIFIED spawn land". That is not the same as {@code select} ignoring these conditions —
+     * every condition {@code select} filters on (quarantine, cooling off, {@code maxLoad} under
+     * every placement policy including the default {@code fixed}, since fleetd #435, model-off,
+     * unreachable, weight-0) is already reflected in the {@link PlacementDecision} this method
+     * returns, because {@code select} walked past every excluded candidate to find it. What this
+     * method's caller must not do is take that resolved name and hand it back to {@link
+     * #spawn(SpawnRequest)} as an explicit profile: the explicit-profile branch treats the same
+     * exclusion conditions as a reason to REFUSE, where {@code select} had already treated them as
+     * a reason to fall through — round 1 of this fix did exactly that, turning a fall-through this
+     * method had already resolved around into a refusal one call later. Round 2 fixes that at the
+     * caller: {@link #spawn(SpawnRequest, PlacementDecision)} carries this exact decision to the
+     * spawn without re-resolving or re-checking it, through the same routing path {@code select}
+     * itself was consulted from.
+     *
+     * @throws PlacementException if no candidate in {@code role}'s pool is currently placeable
+     *                            (mirrors what an actual unqualified spawn would throw)
+     */
+    @Override
+    public PlacementDecision place(MemberRole role) {
+        PlacementContext ctx = placementContextFor(role, new HashSet<>());
+        return new PlacementDecision(placementPolicy.get().select(ctx).profile());
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Delegates to {@link #place}, so the two can never disagree about the answer for the same
+     * {@code role} at the same instant — kept as a convenience for a caller that only wants the
+     * resolved name (a status report, a log line), never for a caller that will act on it by
+     * spawning: that caller must hold the {@link PlacementDecision} itself and pass it to {@link
+     * #spawn(SpawnRequest, PlacementDecision)} — see {@link PlacementDecision}'s javadoc for why
+     * resolving here and spawning separately, with the name fed back in as an explicit profile,
+     * regressed fleetd #425 twice.
+     */
+    @Override
+    public String routedProfileFor(MemberRole role) {
+        return place(role).profile();
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Routes {@code decision.profile()} directly to its owning delegate — the identical
+     * {@code d.spawn(routedReq)} call {@link #spawn(SpawnRequest)}'s blank-profile branch makes for
+     * its first pick — WITHOUT re-running {@link #enforceNotQuarantined}, {@link
+     * #enforceNotCoolingOff}, {@link #enforceMaxLoad}, or {@link #enforceModelEnabled}: those are
+     * the EXPLICIT-profile branch's checks, and {@code decision} did not come from an operator
+     * naming a profile — it came from {@link #place}, which already applied whichever of these
+     * conditions {@link PlacementPolicy#select} actually filters on (fleetd #425 rework, round 2).
+     *
+     * <p>The two branches disagree on purpose about what an excluded profile means, and that
+     * disagreement is not what this method removes. The blank-profile routing branch (and
+     * {@link #place}) treats a quarantined/cooling-off/at-cap/model-off/unreachable/weight-0 profile
+     * as a reason to fall through to the next candidate; the EXPLICIT-profile branch treats naming
+     * that same profile as a reason to refuse outright — someone who names a profile should get a
+     * refusal, not a silent substitution onto a different backend. That is still correct after
+     * fleetd #435. What round 1 got wrong, and what this method exists to stop happening again, is
+     * turning a fall-through into a refusal by accident: resolving a name via {@link #place} and
+     * then handing that same name back to {@link #spawn(SpawnRequest)} as an explicit profile takes
+     * the refusing branch on a decision the routing branch had already approved by falling through
+     * past everything else.
+     *
+     * <p>Before fleetd #435, this exact accident was reachable through {@code maxLoad} specifically:
+     * {@code FixedPlacementPolicy} — the default policy — did not evaluate {@code maxLoad} at all
+     * for automatic selection, so {@link #place} could approve an at-cap profile that {@link
+     * #enforceMaxLoad} would then refuse one call later. fleetd #435 closed that: {@code
+     * FixedPlacementPolicy} now walks past an at-cap candidate exactly like {@code weighted}/
+     * {@code round-robin} already did, so {@link #place} can no longer return one, and this specific
+     * failure — an approved placement dying at {@code enforceMaxLoad} — cannot happen any more.
+     * What this method still buys, now that {@code maxLoad} can no longer cause it: it never
+     * re-evaluates a condition {@link #place} already decided, and it closes the window between
+     * that decision and the spawn in which the underlying state (another spawn landing on the same
+     * profile, a config reload) could otherwise move and make a stale explicit re-check wrong.
+     *
+     * <p>Deliberately does not retry on {@link PeerUnreachableException} across candidates the way
+     * {@link #spawn(SpawnRequest)}'s blank-profile branch does: retrying here would silently
+     * re-place the caller onto a different profile than the one {@code decision} named, behind the
+     * back of a caller that may already have provisioned something (a worktree's {@code repoRoot},
+     * parity overlay) specifically for that name. A caller that wants the composite's own failover
+     * should call {@link #spawn(SpawnRequest)} with a blank profile directly, not resolve through
+     * {@link #place} first. Losing that retry on a resolve-then-spawn path is an accepted, unrelated
+     * cost — see {@code SessionManager.acquireWithWorktree}'s own comment on it — never widened by
+     * this round to include {@code maxLoad}, which is what round 1 actually lost.
+     */
+    @Override
+    public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+        HerdrPeerLauncher d = route(decision.profile());
+        SpawnRequest routedReq = req.withProfile(decision.profile());
+        PeerHandle handle = d.spawn(routedReq);
+        spawnedBy.put(handle.id(), d);
+        return handle;
+    }
+
     @Override
     public String effectiveCwd(SpawnRequest req) {
         return route(req.profileName()).effectiveCwd(req);
@@ -780,9 +920,21 @@ public final class CompositePeerLauncher implements PeerLauncher {
         return byProfile.keySet();
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * <p>fleetd #425: reports the <em>live</em> {@code dev} pool's first entry — the same value
+     * {@link #defaultProfileFor} computes for {@link MemberRole#DEV} — not the {@link
+     * #defaultProfile} field captured at construction. An unqualified {@code fleet_spawn} defaults
+     * to {@code MemberRole#DEV} (see {@link dev.ltms.fleet.peer.SpawnRequest}), so "the dev pool's
+     * live first entry" is exactly the profile such a spawn actually lands on right now — the
+     * question {@code fleet_profiles}' {@code "default"} field exists to answer. The frozen field is
+     * a role-agnostic fallback used only when {@link #poolFor} has nothing to report at all (no
+     * profiles configured), which {@link #defaultProfileFor} already handles.
+     */
     @Override
     public String defaultProfile() {
-        return defaultProfile;
+        return defaultProfileFor(MemberRole.DEV);
     }
 
     /**

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

@@ -17,6 +17,7 @@ import dev.ltms.fleet.peer.PeerHandle;
 import dev.ltms.fleet.peer.PeerLauncher;
 import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.peer.SpawnRequest;
+import dev.ltms.fleet.placement.PlacementDecision;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -594,6 +595,23 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
                 req.sessionName(), spawned.agentSessionId(), spawned.receipt());
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * <p>fleetd #450: re-enters {@link #spawn(SpawnRequest)} with {@code decision}'s profile named
+     * explicitly. This is the re-entering form the interface javadoc describes for a launcher with
+     * no placement concept of its own — an instance of this class spawns a single adapter's own
+     * profile set by explicit name only ({@link #place}/{@link #defaultProfileFor} are unoverridden
+     * here and just wrap {@link #defaultProfile()}); it does no quarantine/cool-off/maxLoad/model-off
+     * filtering of its own to re-apply. That filtering lives one layer up, in {@code
+     * CompositePeerLauncher}, which is the launcher that routes across more than one profile and
+     * therefore overrides this method with the routing form instead.
+     */
+    @Override
+    public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+        return spawn(req.withProfile(decision.profile()));
+    }
+
     /** The herdr daemon that owns this launcher's pane coordinates. */
     public HerdrClient herdr() {
         return agents.herdr();

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

@@ -394,17 +394,17 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     }
 
     @Override
-    public void ack(String target, String msgId) {
+    public boolean ack(String target, String msgId) {
         var perTarget = held.get(target);
         if (perTarget == null || perTarget == RELEASED) {
-            return;
+            return false;
         }
         Held h;
         synchronized (perTarget) {
             h = perTarget.remove(msgId);
         }
         if (h == null) {
-            return;                                // never held (or already acked) — no-op
+            return false;                          // never held (or already acked) — no-op
         }
         try {
             synchronized (channelLock) {
@@ -418,6 +418,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
             }
             throw new IllegalStateException("cannot ack reply " + msgId + " on " + queueName(target), e);
         }
+        return true;
     }
 
     private DeliverCallback deliverCallback(String target) {

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

@@ -59,16 +59,17 @@ public final class InMemoryReplyInbox implements ReplyInbox {
     }
 
     @Override
-    public void ack(String target, String msgId) {
+    public boolean ack(String target, String msgId) {
         if (!owned.contains(target)) {
-            return;
+            return false;
         }
         var perTarget = store.get(target);
-        if (perTarget != null) {
-            //noinspection SynchronizationOnLocalVariableOrMethodParameter
-            synchronized (perTarget) {
-                perTarget.remove(msgId);
-            }
+        if (perTarget == null) {
+            return false;
+        }
+        //noinspection SynchronizationOnLocalVariableOrMethodParameter
+        synchronized (perTarget) {
+            return perTarget.remove(msgId) != null;
         }
     }
 }

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

@@ -826,9 +826,13 @@ public final class MessageService {
     /**
      * Acknowledge a specific reply by {@code msgId} for {@code target}. Removes it from the inbox
      * so that a subsequent drain or peek no longer returns it.
+     *
+     * @return {@code true} if an entry was actually removed, {@code false} if {@code msgId} was not
+     *     in {@code target}'s inbox (wrong id, wrong target, or already acked). The caller —
+     *     {@link dev.ltms.fleet.mcp.FleetMcp#ack} — must not report success on {@code false}.
      */
-    public void ackReply(String target, String msgId) {
-        inbox.ack(target, msgId);
+    public boolean ackReply(String target, String msgId) {
+        return inbox.ack(target, msgId);
     }
 
     /**

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

@@ -46,6 +46,13 @@ public interface ReplyInbox {
     /** Non-destructive snapshot of pending replies for {@code target} (FIFO), empty list if none. */
     List<InboxMessage> peek(String target);
 
-    /** Remove the reply {@code msgId} for {@code target} once the primary has taken it. No-op if absent. */
-    void ack(String target, String msgId);
+    /**
+     * Remove the reply {@code msgId} for {@code target} once the primary has taken it.
+     *
+     * @return {@code true} if an entry was actually removed, {@code false} if there was nothing to
+     *     remove (unknown {@code target}, unowned {@code target}, or a {@code msgId} not held for
+     *     it). A {@code false} is not an error — acking a {@code target} this daemon does not own is
+     *     part of the normal contract, not a failure.
+     */
+    boolean ack(String target, String msgId);
 }

fleetd/src/main/java/dev/ltms/fleet/peer/PeerLauncher.java

@@ -1,5 +1,7 @@
 package dev.ltms.fleet.peer;
 
+import dev.ltms.fleet.placement.PlacementDecision;
+
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Set;
@@ -143,9 +145,136 @@ public interface PeerLauncher {
 
     /**
      * The profile a no-argument {@link #spawn(SpawnRequest)} uses, or {@code null} if none is configured.
+     *
+     * <p>fleetd #425: for an implementation with role pools (a no-argument spawn is read as {@link
+     * MemberRole#DEV}, see {@link SpawnRequest}), this must be the profile a live spawn of that role
+     * would actually be placed on right now, not a value captured once at startup — a caller such as
+     * {@code fleet_profiles} relies on this to report a live, not frozen, fact.
      */
     String defaultProfile();
 
+    /**
+     * The profile an unqualified spawn of {@code role} would resolve to right now — the role-aware,
+     * live counterpart of {@link #defaultProfile()} (fleetd #425).
+     *
+     * <p>A caller that must provision something profile-specific (working directory, parity overlay
+     * files) <em>before</em> the actual spawn — {@code SessionManager.acquireWithWorktree} is the one
+     * that exists today — needs the exact profile that spawn will use, for the caller's real role,
+     * not a role-agnostic guess. Calling {@link #defaultProfile()} for that purpose reads {@code
+     * MemberRole#DEV}'s answer regardless of the caller's actual role, which is wrong for any other
+     * role and can provision for a profile the spawn never lands on.
+     *
+     * <p>Default implementation returns {@link #defaultProfile()}, ignoring {@code role} — the right
+     * answer for a launcher with no role-pool concept of its own (e.g. a single {@code
+     * HerdrPeerLauncher} adapter, which is never reached this way in production: {@code
+     * CompositePeerLauncher} always fronts it and resolves roles itself).
+     */
+    default String defaultProfileFor(MemberRole role) {
+        return defaultProfile();
+    }
+
+    /**
+     * The profile an <em>unqualified</em> spawn of {@code role} would actually be routed to right
+     * now — the same candidate list, the same {@code quarantined}/{@code coolingOff}/{@code
+     * modelOff} filtering, and the same {@code PlacementPolicy} that {@link #spawn} itself
+     * consults for a blank-profile request (fleetd #425 rework).
+     *
+     * <p>This is <em>not</em> {@link #defaultProfileFor}: that method answers "what is first in
+     * {@code role}'s pool", blind to quarantine, cool-off, and the model on/off gate — the right
+     * answer for a role-agnostic, best-effort report ({@code fleet_profiles}' {@code "default"}
+     * field), but the wrong one for a caller that needs the profile a spawn will actually land on.
+     * A quarantined or model-off pool-first profile makes {@link #defaultProfileFor} return a name
+     * an unqualified spawn will never be routed to.
+     *
+     * <p>Just the resolved name, not the full {@link PlacementDecision} — a caller that only wants
+     * to know the answer (a status report, a log line) can call this; a caller that will later
+     * <em>act</em> on the answer by spawning — provisioning a worktree for a specific profile
+     * before the peer exists is the one that matters — must call {@link #place} and carry the
+     * {@link PlacementDecision} itself through to {@link #spawn(SpawnRequest, PlacementDecision)}
+     * instead of calling this method and feeding the string back in as an explicit profile. Doing
+     * that re-enters {@link #spawn(SpawnRequest)}'s explicit-profile branch, which disagrees with
+     * the routing branch on purpose about what an excluded profile means: the routing branch (and
+     * {@link #place}) falls through a quarantined/cooling-off/at-cap/model-off/unreachable/weight-0
+     * profile to the next candidate, while the explicit branch refuses outright — correct for an
+     * operator who named that profile on purpose, wrong for a name that only ever came from placement
+     * itself. That accidental refusal is exactly the regression fleetd #425 rework round 2 fixes:
+     * the default implementation below delegates to {@link #place}, so the two can never drift apart,
+     * but a caller that resolves through this method alone and spawns separately can still recreate
+     * the round-1 defect for itself. (Before fleetd #435, this accident was also reachable through
+     * {@code maxLoad} specifically, because {@code FixedPlacementPolicy} — the default policy — did
+     * not evaluate it at all for automatic selection; #435 closed that gap, so a placement decision
+     * can no longer be at cap in the first place. The refusal-vs-fall-through disagreement above is
+     * the part that was never about {@code maxLoad} and is still real.)
+     *
+     * @throws RuntimeException (implementation-specific, typically a placement exception) if no
+     *                          candidate in {@code role}'s pool is currently placeable
+     */
+    default String routedProfileFor(MemberRole role) {
+        return place(role).profile();
+    }
+
+    /**
+     * Resolve, <em>without spawning</em>, the {@link PlacementDecision} an unqualified spawn of
+     * {@code role} would make right now — the same candidate list, the same {@code
+     * quarantined}/{@code coolingOff}/{@code modelOff} filtering, and the same {@code
+     * PlacementPolicy} {@link #spawn(SpawnRequest)}'s blank-profile branch itself consults (fleetd
+     * #425 rework).
+     *
+     * <p>Pair this with {@link #spawn(SpawnRequest, PlacementDecision)}, never with {@link
+     * #spawn(SpawnRequest)} fed the decision's profile as an explicit name — see {@link
+     * PlacementDecision}'s own javadoc for why that second form regressed.
+     *
+     * <p>Default implementation wraps {@link #defaultProfile()}, ignoring {@code role} and every
+     * placement condition — the right answer for a launcher with no pool or placement-policy
+     * concept of its own, matching {@link #defaultProfileFor}'s own default.
+     *
+     * @throws RuntimeException (implementation-specific, typically a placement exception) if no
+     *                          candidate in {@code role}'s pool is currently placeable
+     */
+    default PlacementDecision place(MemberRole role) {
+        return new PlacementDecision(defaultProfile());
+    }
+
+    /**
+     * Spawn against an already-resolved {@link PlacementDecision} from {@link #place}, honoring it
+     * completely: none of the conditions {@link #place} already applied — quarantine, cooling off,
+     * {@code maxLoad} (evaluated by every placement policy including the default {@code fixed},
+     * since fleetd #435), model-off — are re-evaluated here; {@code decision} already reflects them.
+     * This is not skipping a check {@code place} left undone; it is not repeating one {@code place}
+     * already did, and not re-opening the window between that decision and this spawn in which the
+     * underlying state could otherwise move. This is what lets a resolve-then-spawn caller
+     * ({@code SessionManager.acquireWithWorktree}, which must know the profile before it can
+     * provision a worktree for it) and a plain blank-profile {@link #spawn(SpawnRequest)} caller
+     * land on the exact same outcome for the exact same placement state (fleetd #425 rework,
+     * round 2).
+     *
+     * <p>{@code req}'s own {@link SpawnRequest#profileName()} is ignored in favor of {@code
+     * decision.profile()} — the caller is expected to have built {@code req} with a blank or
+     * matching profile; passing a request that names a <em>different</em>, explicit profile than
+     * the decision it is paired with is a caller bug this method does not attempt to detect.
+     *
+     * <p>No default implementation (fleetd #450): the two correct bodies disagree on purpose, so an
+     * implementer must choose one rather than silently inherit whichever this interface happened to
+     * provide. An implementer with no placement concept of its own — spawns a single profile, e.g.
+     * {@code HerdrPeerLauncher} — should delegate to {@link #spawn(SpawnRequest)} with the decision's
+     * profile named explicitly, since there is no separate routing path to honor there: the
+     * explicit-profile branch it re-enters and the routing branch {@link #place} would have used are
+     * the same thing. <strong>A launcher that routes across more than one profile — the way {@code
+     * CompositePeerLauncher} routes across every configured adapter — MUST NOT re-enter {@link
+     * #spawn(SpawnRequest)}.</strong> Doing so re-applies that single-argument method's
+     * explicit-profile checks ({@code enforceNotQuarantined}, {@code enforceNotCoolingOff}, {@code
+     * enforceMaxLoad}, {@code enforceModelEnabled} in {@code CompositePeerLauncher}), which can
+     * refuse the very profile {@link #place} just chose, if the underlying placement state moved in
+     * the window between the {@link #place} call and this one — the exact window this method and
+     * {@link PlacementDecision} exist to close (fleetd #444). Before #450 this was a {@code default}
+     * method that only {@code CompositePeerLauncher} overrode; a future placement-doing launcher
+     * could have inherited the re-entering body silently and never known. Making it abstract turns
+     * that silent inheritance into a compile error.
+     *
+     * @throws IllegalArgumentException if the decision names an unknown profile
+     */
+    PeerHandle spawn(SpawnRequest req, PlacementDecision decision);
+
     /**
      * Resolve the effective working directory for a spawn {@code req} without actually spawning.
      * Resolution order: requestedCwd → profile cwd → callerCwd → daemon cwd.

fleetd/src/main/java/dev/ltms/fleet/placement/PlacementDecision.java

@@ -0,0 +1,49 @@
+package dev.ltms.fleet.placement;
+
+import dev.ltms.fleet.peer.MemberRole;
+import dev.ltms.fleet.peer.PeerLauncher;
+import dev.ltms.fleet.peer.SpawnRequest;
+
+/**
+ * An already-completed placement choice — the outcome of one {@link PeerLauncher#place} call,
+ * carried forward so a later {@link PeerLauncher#spawn(SpawnRequest, PlacementDecision)} can honor
+ * it directly instead of re-resolving the profile a second time (fleetd #425 rework, round 2).
+ *
+ * <p>The problem this exists to close: a caller that must know the profile <em>before</em> it can
+ * spawn — {@code SessionManager.acquireWithWorktree} provisions a worktree's {@code repoRoot} and
+ * parity overlay for a specific profile before the peer process exists — used to resolve that name
+ * with {@code PeerLauncher.routedProfileFor(role)} and then hand the SAME string back to {@link
+ * PeerLauncher#spawn(SpawnRequest)} as an EXPLICIT profile. That re-resolution is not free: naming
+ * a profile explicitly makes {@code CompositePeerLauncher.spawn} take its THROWING branch
+ * ({@code enforceNotQuarantined}/{@code enforceNotCoolingOff}/{@code enforceMaxLoad}/{@code
+ * enforceModelEnabled}), while an unqualified spawn's ROUTING branch never runs those checks at
+ * all — it instead FALLS THROUGH to the next candidate on exactly the same conditions the throwing
+ * branch refuses on. That disagreement is deliberate: an operator who names a profile should get a
+ * refusal, not a silent substitution. The bug is turning the fall-through into a refusal by
+ * accident — resolving a name through the routing side and then re-entering the refusing side with
+ * it, for a decision the routing side had already approved by walking past everything else.
+ * Before fleetd #435, this accident was also reachable through {@code maxLoad} specifically: the
+ * default {@code fixed} placement policy did not evaluate {@code maxLoad} at all for automatic
+ * selection, so a profile placement itself just approved could still die at {@code enforceMaxLoad}
+ * one call later, purely because the caller's route to the spawn passed through an explicit
+ * profile name instead of the routing branch — a failure a worktree-less unqualified spawn would
+ * never hit. fleetd #435 closed that specific gap ({@code fixed} now evaluates {@code maxLoad}
+ * exactly like every other placement policy), so a {@link PlacementDecision} can no longer be
+ * at-cap in the first place — but the refusal-vs-fall-through disagreement above was never about
+ * {@code maxLoad}, and resolving a name and re-entering the refusing branch with it is still wrong
+ * for every OTHER condition placement filters on.
+ *
+ * <p>{@link PeerLauncher#spawn(SpawnRequest, PlacementDecision)} closes that by spawning through
+ * the identical code path the routing branch itself uses, keyed off the SAME decision {@link
+ * PeerLauncher#place} returned — no re-checking of any condition placement already evaluated. A
+ * resolve-then-spawn caller and a blank-profile {@link PeerLauncher#spawn(SpawnRequest)} caller can
+ * then never disagree about which conditions apply to the same placement state, and neither one
+ * re-opens the window between the placement decision and the spawn in which the underlying state
+ * could otherwise move.
+ *
+ * @param profile the profile this decision resolved to (may be {@code null} only when no profile is
+ *                configured at all — the same corner case {@link PeerLauncher#defaultProfile()}
+ *                already tolerates)
+ */
+public record PlacementDecision(String profile) {
+}

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

@@ -10,6 +10,7 @@ import dev.ltms.fleet.peer.MemberRole;
 import dev.ltms.fleet.peer.PeerHandle;
 import dev.ltms.fleet.peer.PeerLauncher;
 import dev.ltms.fleet.peer.SpawnRequest;
+import dev.ltms.fleet.placement.PlacementDecision;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -584,8 +585,65 @@ public final class SessionManager implements TurnListener {
                                               String ownerTerminal, WorktreeRequest wt,
                                                String sessionName, String resumeSessionId,
                                                MemberLifecycle.SlotReservation reservation) {
-        String preResolvedProfile = (profile == null || profile.isBlank())
-                ? launcher.defaultProfile() : profile;
+        // fleetd #425 rework (round 2): resolved through launcher.place(memberRole) — the same
+        // candidate list, quarantine/cool-off/model-off filtering, and PlacementPolicy an unqualified
+        // spawn of this role is actually judged against right now — never launcher.defaultProfile()
+        // (MemberRole.DEV only, wrong for any other role) and never launcher.defaultProfileFor()
+        // (the role's pool FIRST entry, blind to quarantine/cool-off/model-off: a first-round fix
+        // used exactly this and regressed fleetd #429's "the fleet keeps working when a model is
+        // turned off" guarantee — a quarantined or model-off pool-first profile made this throw
+        // instead of routing around it, which an unqualified spawn is supposed to do). This same
+        // resolved decision is reused below for repoRoot, parityOverlay, AND the spawn itself so the
+        // worktree is always provisioned for the profile the member actually runs on — the two could
+        // disagree before fleetd #425: this name picked repoRoot/overlay, but the spawn below passed
+        // the ORIGINAL (blank) profile through to placement, which re-resolves live and can pick a
+        // different profile if the pool changed between the two reads, or a genuinely different one
+        // under weighted/round-robin placement.
+        //
+        // Round 1 of this rework fed the resolved name back into launcher.spawn(SpawnRequest) as an
+        // EXPLICIT profile. That was a mistake this round corrects, and the mistake is not that the
+        // two branches apply different checks — they are SUPPOSED to disagree: the routing branch a
+        // blank spawn takes treats a quarantined/cooling-off/at-cap/model-off/unreachable/weight-0
+        // profile as a reason to fall through to the next candidate, while CompositePeerLauncher's
+        // THROWING branch (enforceNotQuarantined/enforceNotCoolingOff/enforceMaxLoad/
+        // enforceModelEnabled) treats naming that same profile explicitly as a reason to refuse
+        // outright. That is correct: an operator who names a profile should get a refusal, not a
+        // silent substitution onto a different backend. The mistake was turning a fall-through into
+        // a refusal by accident — resolving a name via the routing side and then re-entering the
+        // refusing side with it, for a placement the routing side had already approved by walking
+        // past everything else.
+        //
+        // Before fleetd #435, this accident was reachable through maxLoad specifically: the default
+        // `fixed` placement policy did not evaluate maxLoad at all for automatic selection, so an
+        // at-cap pool-first profile that placement itself would have picked for a plain unqualified
+        // spawn could die at enforceMaxLoad one call later, purely because this method's route to
+        // the spawn passed through an explicit profile name — a failure a worktree-less unqualified
+        // spawn never hit. fleetd #435 closed that gap (`fixed` now evaluates maxLoad exactly like
+        // every other placement policy), so that specific failure can no longer happen — a
+        // PlacementDecision this method resolves can no longer be at-cap in the first place. What
+        // this round's fix still buys, now that maxLoad can no longer cause the accident: it keeps
+        // the PlacementDecision from place() and hands it to launcher.spawn(SpawnRequest,
+        // PlacementDecision) for an unqualified request, which spawns through the SAME routing
+        // branch a blank spawn uses — no enforce* check is newly applied, and the window between the
+        // placement decision and the spawn (in which the pool, a config reload, or another spawn
+        // landing on the same profile could otherwise move the state) never reopens. An
+        // explicitly-named profile still goes through launcher.spawn(SpawnRequest) and its throwing
+        // branch, unchanged — that caller asked for one profile by name and still gets everything
+        // enforceNotQuarantined/enforceNotCoolingOff/enforceMaxLoad/enforceModelEnabled decide about
+        // it, refusal included.
+        //
+        // The one cost that remains, unchanged from round 1: an unqualified worktree-provisioned
+        // spawn does not get CompositePeerLauncher's cross-candidate retry on a live
+        // PeerUnreachableException raised by the backend itself at spawn time (a transport-level
+        // failure placement cannot see in advance) — spawn(req, decision) commits to the one profile
+        // place() already chose, the same way an explicit-profile spawn commits to its one name. That
+        // trade is deliberate: a worktree provisioned for the wrong backend (the #425 hazard) is worse
+        // than a spawn that fails cleanly and can be retried by the caller. Nothing else is lost:
+        // maxLoad, quarantine, cool-off and model-off all behave identically whether or not a
+        // worktree was requested — that agreement is the invariant this rework exists to hold.
+        boolean unqualifiedProfile = profile == null || profile.isBlank();
+        PlacementDecision decision = unqualifiedProfile ? launcher.place(memberRole) : new PlacementDecision(profile);
+        String preResolvedProfile = decision.profile();
         // CB-507: resolve through the launcher's CB-112 chain (requested → profile cwd → caller →
         // daemon cwd → "."), never the raw args. A plain REST spawn supplies neither a requested
         // nor a caller cwd, so taking the first non-blank of those two yielded null and put
@@ -608,7 +666,15 @@ public final class SessionManager implements TurnListener {
             // copies more files into the worktree after add() returns, so sharing the group any earlier
             // leaves those overlay files operator-owned and read-only for a different-uid member.
             worktrees.shareWithGroup(repoRoot, path);
-            handle = launcher.spawn(new SpawnRequest(profile, path, callerCwd, sessionName, resumeSessionId, memberRole));
+            // fleetd #425: preResolvedProfile, not the original (possibly blank) profile — see the
+            // comment above where it is resolved. The overlay/repoRoot above and the spawn here must
+            // name the same profile. An unqualified request stays unqualified here and is honored via
+            // the PlacementDecision already captured above (spawn(req, decision) — the routing branch,
+            // no enforce* re-check); an explicitly-named profile still goes through the single-arg
+            // spawn(req) and its throwing branch, exactly as before this rework.
+            SpawnRequest spawnReq = new SpawnRequest(unqualifiedProfile ? null : preResolvedProfile,
+                    path, callerCwd, sessionName, resumeSessionId, memberRole);
+            handle = unqualifiedProfile ? launcher.spawn(spawnReq, decision) : launcher.spawn(spawnReq);
         } catch (RuntimeException e) {
             log.warn("spawn failed for profile={} role={} branch={} path={}: {}",
                     preResolvedProfile, memberRole, branch, path, e.getMessage());
@@ -636,7 +702,7 @@ public final class SessionManager implements TurnListener {
             }
             throw e;
         }
-        String resolvedProfile = resolveProfile(handle, profile);
+        String resolvedProfile = resolveProfile(handle, preResolvedProfile);
         String cwd = launcher.effectiveCwd(new SpawnRequest(resolvedProfile, path, callerCwd));
         long now = nowNanos.getAsLong();
         // CB-619: see the no-worktree path above — bind before recording, and store the returned

fleetd/src/test/java/dev/ltms/fleet/FleetdBackendErrorSinkTest.java

@@ -20,6 +20,7 @@ import dev.ltms.fleet.peer.PeerLauncher;
 import dev.ltms.fleet.peer.SpawnRequest;
 import dev.ltms.fleet.placement.BackendOutagePolicy;
 import dev.ltms.fleet.placement.BackendQuarantine;
+import dev.ltms.fleet.placement.PlacementDecision;
 import dev.ltms.fleet.placement.PlacementPolicies;
 import dev.ltms.fleet.session.MemberSession;
 import dev.ltms.fleet.session.SessionManager;
@@ -123,6 +124,11 @@ class FleetdBackendErrorSinkTest {
             throw new UnsupportedOperationException("not reachable — this test never acquires a session");
         }
 
+        @Override
+        public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+            throw new UnsupportedOperationException("not reachable — this test never acquires a session");
+        }
+
         @Override
         public Set<String> profiles() {
             return Set.of();

fleetd/src/test/java/dev/ltms/fleet/FleetdExhaustionDetectionArmedWiringTest.java

@@ -2,6 +2,7 @@ package dev.ltms.fleet;
 
 import dev.ltms.fleet.config.ConfigRef;
 import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.inject.LiveExhaustedPatterns;
 import dev.ltms.fleet.mcp.FleetMcp;
 import dev.ltms.fleet.placement.BackendQuarantine;
 import org.junit.jupiter.api.DisplayName;
@@ -11,16 +12,31 @@ import org.junit.jupiter.api.io.TempDir;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Map;
-import java.util.regex.Pattern;
+
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
- * fleetd #404: {@code exhaustionDetectionArmed} must describe the startup pattern map, not the
- * reloaded config snapshot.
+ * fleetd #446: {@code exhaustionDetectionArmed} must describe the LIVE config, not a startup
+ * snapshot — the opposite of what fleetd #404 asked for. #404 pinned {@code exhaustedPattern} as
+ * compiled once at startup (matching {@code CompletionResolver}'s then-frozen behaviour); #446's
+ * whole point is that both sides — the classification {@code CompletionResolver} enforces and the
+ * {@code exhaustionDetectionArmed} field this reports — now read the SAME live {@link
+ * LiveExhaustedPatterns} instance, so a reload arms or disarms detection with no restart, and the
+ * report can never disagree with the behaviour (the fleetd #404 rule, now upheld for real instead
+ * of by freezing both sides).
  *
  * <p>This test needs a reload. At startup the two snapshots agree, so a test of only a newly
  * started daemon would not detect a live {@code config.get()} lookup in the report field.
+ *
+ * <p><b>The hotness proof criterion 1 demands:</b> run this against the fixed code (green) and
+ * against the pre-#446 compile site — {@code Fleetd.main}'s {@code exhaustedPatternsByProfile}
+ * compiled once into a {@code Map<String, Pattern>} at startup, handed to {@code quarantineSource}
+ * — and show it fails there. See the class doc history above: that shape is exactly what
+ * {@code reloadedPatternArmsDetectionWithNoRestart} below is written to catch, and it is the test
+ * that used to assert the opposite (see git history for this file's pre-#446 version, which
+ * asserted {@code assertFalse(...)} on the identical scenario this now asserts {@code assertTrue}
+ * on).
  */
 class FleetdExhaustionDetectionArmedWiringTest {
 
@@ -54,41 +70,68 @@ class FleetdExhaustionDetectionArmedWiringTest {
             """;
 
     @Test
-    @DisplayName("reloading an exhaustedPattern does not arm the startup detection source")
-    void reloadedPatternDoesNotChangeTheArmedFieldUntilRestart(@TempDir Path dir) throws Exception {
+    @DisplayName("reloading a profile's exhaustedPattern IN arms exhaustionDetectionArmed, no restart")
+    void reloadedPatternArmsDetectionWithNoRestart(@TempDir Path dir) throws Exception {
         Path file = dir.resolve("fleetd.yaml");
         Files.writeString(file, NO_PATTERN);
         ConfigRef config = new ConfigRef(file, FleetConfig.load(file));
 
+        FleetMcp.QuarantineSource before = Fleetd.quarantineSource(config, BackendQuarantine.none(),
+                new LiveExhaustedPatterns(() -> config.get().profiles()), Map.of());
+        assertFalse(before.exhaustedPatternArmed().apply("terra"),
+                "no exhaustedPattern configured yet — must report unarmed");
+
         Files.writeString(file, WITH_PATTERN);
         assertTrue(config.reload().applied());
         assertTrue(config.get().profiles().get("terra").hasExhaustedPattern());
 
-        FleetMcp.QuarantineSource source = Fleetd.quarantineSource(config, BackendQuarantine.none(),
-                Map.of());
-        assertFalse(source.exhaustedPatternArmed().apply("terra"),
-                "exhaustionDetectionArmed must use the startup pattern map, not config.get()");
+        // Re-read exhaustedPatternArmed off the SAME QuarantineSource built BEFORE the reload — no
+        // new object, no new wiring — to prove the field itself is live, not merely that a freshly
+        // built source would be. This is the exact axis the pre-#446 code fails on: the same
+        // Function<String,Boolean> lambda, called again after a reload, sees the new answer only if
+        // it re-reads config.get() on every call rather than a value captured earlier.
+        assertTrue(before.exhaustedPatternArmed().apply("terra"),
+                "exhaustionDetectionArmed must read the LIVE config on every call — fleetd #446 made "
+                        + "this hot; it must reflect a reload with no restart and no new QuarantineSource");
     }
 
     @Test
-    @DisplayName("a profile in the startup pattern map is reported as armed")
-    void aProfileInTheStartupMapIsArmed(@TempDir Path dir) throws Exception {
-        // fleetd #404, second direction. The test above only ever passes an EMPTY startup map, so
-        // it cannot tell a correct lookup from one that is permanently off. Measured: replacing the
-        // armed lambda with `profile -> false` left the whole suite green at 1475 tests. That
-        // mutation would make #395's visibility feature dead — an operator fixing a detection gap
-        // would be told the gap is still open after fixing it, forever. Both directions are needed:
-        // this test is the only thing that fails when the field stops reporting armed at all.
+    @DisplayName("reloading exhaustedPattern OUT disarms exhaustionDetectionArmed, no restart")
+    void reloadedPatternRemovalDisarmsDetectionWithNoRestart(@TempDir Path dir) throws Exception {
         Path file = dir.resolve("fleetd.yaml");
         Files.writeString(file, WITH_PATTERN);
         ConfigRef config = new ConfigRef(file, FleetConfig.load(file));
 
         FleetMcp.QuarantineSource source = Fleetd.quarantineSource(config, BackendQuarantine.none(),
-                Map.of("terra", Pattern.compile("usage limit")));
+                new LiveExhaustedPatterns(() -> config.get().profiles()), Map.of());
+        assertTrue(source.exhaustedPatternArmed().apply("terra"),
+                "exhaustedPattern is configured from the start — must report armed");
+
+        Files.writeString(file, NO_PATTERN);
+        assertTrue(config.reload().applied());
+
+        assertFalse(source.exhaustedPatternArmed().apply("terra"),
+                "removing exhaustedPattern and reloading must disarm detection with no restart — an "
+                        + "operator turning detection off (e.g. while debugging a false positive) "
+                        + "must see that reflected immediately, exactly like arming it is");
+    }
+
+    @Test
+    @DisplayName("a profile with no exhaustedPattern configured is reported unarmed")
+    void aProfileWithNoPatternIsUnarmed(@TempDir Path dir) throws Exception {
+        // fleetd #404's second direction, carried forward: this test alone fails if
+        // exhaustedPatternArmed degenerates to a constant `profile -> true` — it needs at least one
+        // profile that is genuinely unarmed to catch that.
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, WITH_PATTERN);
+        ConfigRef config = new ConfigRef(file, FleetConfig.load(file));
+
+        FleetMcp.QuarantineSource source = Fleetd.quarantineSource(config, BackendQuarantine.none(),
+                new LiveExhaustedPatterns(() -> config.get().profiles()), Map.of());
 
         assertTrue(source.exhaustedPatternArmed().apply("terra"),
-                "a profile whose pattern was compiled at startup must report armed");
+                "a profile whose exhaustedPattern is configured must report armed");
         assertFalse(source.exhaustedPatternArmed().apply("sonnet"),
-                "a profile absent from the startup map must not report armed");
+                "a profile absent from config entirely must not report armed");
     }
 }

fleetd/src/test/java/dev/ltms/fleet/FleetdExhaustionSinkWarningTest.java

@@ -0,0 +1,175 @@
+package dev.ltms.fleet;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.WorkspaceControl;
+import dev.ltms.fleet.inject.ExhaustionSink;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.placement.BackendQuarantine;
+import dev.ltms.fleet.session.SessionManager;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #446 follow-up (round 3): round 2's {@link FleetdUsageLimitFixWarningTest} calls {@link
+ * Fleetd#usageLimitFixWarning}/{@link Fleetd#usageLimitFixWarningNoModel} directly, which pins
+ * the two methods' TEXT but cannot see whether {@link Fleetd#exhaustionSink}'s {@code log.warn}
+ * call actually invokes either one, or invokes the right one. A mutation battery run against
+ * round 2's merge (1592 green tests) proved both gaps real:
+ *
+ * <ul>
+ *   <li><b>Cell A</b> — replacing the whole ternary result with a literal string
+ *       ({@code "usage-limit fix: MUTANT"}) left every test green;</li>
+ *   <li><b>Cell B</b> — swapping the ternary's two branches, so a profile WITH a {@code model:}
+ *       gets the no-model fallback message and vice versa, was measured unmeasured by the lead
+ *       but the existing test file shows no call that would notice it either.</li>
+ * </ul>
+ *
+ * <p>This class drives {@link Fleetd#exhaustionSink} — the actual factory {@code main} wires,
+ * since round 3's extraction pulled it out of the inline lambda for exactly this reason — and
+ * asserts on the REAL text a {@link ListAppender} attached to {@code Fleetd}'s own logger
+ * captures, mirroring {@code ExhaustedPatternGapReportTest}'s idiom. Chosen over a source-text
+ * assertion (the {@code FleetMcpAuthzTest} {@code everyRegisteredToolHasItsHandlerActionPinned}
+ * idiom) because that route can prove Cell A (the ternary is still there, still built from the
+ * two named methods) but not Cell B (which branch a given profile actually reaches) — this route
+ * answers both from one mechanism, since it reads what the sink actually logged for each shape of
+ * profile.
+ *
+ * <p>Each test filters for the WARN line starting {@code "usage-limit fix:"} specifically — {@link
+ * Fleetd#exhaustionSink} also logs a separate {@code "credential '...' quarantined for..."} WARN
+ * on every call, and asserting against the wrong one would pass or fail for the wrong reason. That
+ * prefix survives even under the M5 mutation above (the mutant keeps the tag, only the body
+ * becomes {@code "MUTANT"}), so the filter itself is not what either mutation defeats.
+ */
+class FleetdExhaustionSinkWarningTest {
+
+    private static final String YAML = """
+            bind:
+              host: 127.0.0.1
+              port: 8765
+            herdrSocket: ~/.config/herdr/herdr.sock
+            profiles:
+              terra:
+                baseUrl: http://gx00.gw:8000
+                model: claude-opus-5
+              gx:
+                baseUrl: http://gx00.gw:8000
+            guard:
+              offSubscriptionHosts:
+                - gx00.gw
+            """;
+
+    private static SessionManager emptyRosterSessions() {
+        FakeHerdr h = new FakeHerdr();
+        FleetConfig.Profile dummy = new FleetConfig.Profile(
+                "dummy", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher launcher = new ClaudeCodeLauncher(new AgentControl(h), new WorkspaceControl(h),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(dummy.profile(), dummy), dummy.profile(), _ -> "tok");
+        // Never acquires a session — exhaustionSink's roster lookup is expected to miss and fall
+        // back to the profileHint argument, exactly like OpenCodeLauncher's real call site does
+        // (fleetd #234) — so the sink under test never needs a populated roster.
+        return new SessionManager(launcher);
+    }
+
+    private static ListAppender<ILoggingEvent> attach() {
+        Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        return appender;
+    }
+
+    private static void detach(ListAppender<ILoggingEvent> appender) {
+        ((Logger) LoggerFactory.getLogger(Fleetd.class)).detachAppender(appender);
+    }
+
+    /** The one WARN line {@code exhaustionSink} builds from the ternary under test, or {@code null}. */
+    private static String fixWarning(ListAppender<ILoggingEvent> appender) {
+        List<String> warns = appender.list.stream()
+                .filter(e -> e.getLevel() == Level.WARN)
+                .map(ILoggingEvent::getFormattedMessage)
+                .filter(m -> m.startsWith("usage-limit fix:"))
+                .toList();
+        assertTrue(warns.size() == 1,
+                "expected exactly one 'usage-limit fix:' WARN per onExhausted call, got " + warns.size()
+                        + ": " + warns);
+        return warns.getFirst();
+    }
+
+    @Test
+    @DisplayName("a profile WITH a configured model gets the actionable models.allow fix, not the fallback")
+    void profileWithModelGetsTheActionableFix(@TempDir Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, YAML);
+        FleetConfig cfg = FleetConfig.load(file);
+        ConfigRef config = ConfigRef.fixed(cfg);
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        Map<String, String> reasonByCredential = new HashMap<>();
+        ExhaustionSink sink = Fleetd.exhaustionSink(emptyRosterSessions(), config, quarantine,
+                reasonByCredential, cfg);
+
+        ListAppender<ILoggingEvent> appender = attach();
+        try {
+            sink.onExhausted("term_x", "The usage limit has been reached", "terra");
+        } finally {
+            detach(appender);
+        }
+
+        String warning = fixWarning(appender);
+        assertTrue(warning.contains("terra"), "must name the profile: " + warning);
+        assertTrue(warning.contains("claude-opus-5"), "must name the model: " + warning);
+        assertTrue(warning.contains("enabled: false"), "must name the action: " + warning);
+        assertTrue(warning.contains("models.allow"), "must name where the action goes: " + warning);
+        assertFalse(warning.contains("has no model: configured"),
+                "a profile WITH a model must not get the no-model fallback text: " + warning);
+    }
+
+    @Test
+    @DisplayName("a profile with NO configured model gets the fallback, never a fix that does not exist")
+    void profileWithNoModelGetsTheFallbackNotAFix(@TempDir Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, YAML);
+        FleetConfig cfg = FleetConfig.load(file);
+        ConfigRef config = ConfigRef.fixed(cfg);
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        Map<String, String> reasonByCredential = new HashMap<>();
+        ExhaustionSink sink = Fleetd.exhaustionSink(emptyRosterSessions(), config, quarantine,
+                reasonByCredential, cfg);
+
+        ListAppender<ILoggingEvent> appender = attach();
+        try {
+            sink.onExhausted("term_y", "The usage limit has been reached", "gx");
+        } finally {
+            detach(appender);
+        }
+
+        String warning = fixWarning(appender);
+        assertTrue(warning.contains("gx"), "must name the profile: " + warning);
+        assertTrue(warning.contains("has no model: configured"),
+                "must say there is no model to gate by name: " + warning);
+        assertFalse(warning.contains("enabled: false"),
+                "a profile with no model: configured has no models.allow entry to flip — must "
+                        + "not claim one exists: " + warning);
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdStartupReportTest.java

@@ -0,0 +1,79 @@
+package dev.ltms.fleet;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.ltms.fleet.config.FleetConfig;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Proves {@link Fleetd#main(String[])} calls every startup report before validation aborts startup.
+ * The invalid non-loopback bind makes {@link FleetConfig#validateAll()} throw before {@code main}
+ * can open the herdr socket or bind a port. The fixture also triggers every report, so removing any
+ * one call from {@code main} leaves its expected log line absent.
+ */
+class FleetdStartupReportTest {
+
+    private static Level originalLevel;
+
+    private static ListAppender<ILoggingEvent> attach() {
+        Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
+        originalLevel = logger.getLevel();
+        logger.setLevel(Level.INFO);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        return appender;
+    }
+
+    private static void detach(ListAppender<ILoggingEvent> appender) {
+        Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
+        logger.detachAppender(appender);
+        logger.setLevel(originalLevel);
+    }
+
+    private static boolean contains(ListAppender<ILoggingEvent> appender, String fragment) {
+        return appender.list.stream()
+                .map(ILoggingEvent::getFormattedMessage)
+                .anyMatch(message -> message.contains(fragment));
+    }
+
+    @Test
+    void mainReportsEveryStartupGapBeforeValidationAborts(@TempDir Path dir) throws Exception {
+        Path config = dir.resolve("fleetd.yaml");
+        Files.writeString(config, """
+                bind:
+                  host: 0.0.0.0
+                  port: 8765
+                profiles:
+                  worker:
+                    baseUrl: https://llm.ltms.dev/v1
+                    gitTokenEnv: GITEA_TOKEN
+                """);
+
+        ListAppender<ILoggingEvent> appender = attach();
+        try {
+            assertThrows(IllegalStateException.class, () -> Fleetd.main(new String[]{config.toString()}));
+        } finally {
+            detach(appender);
+        }
+
+        assertTrue(contains(appender, "startup git host GITEA_HOST:"),
+                "Fleetd.main must report the git host shape");
+        assertTrue(contains(appender, "member trust model: members run as the same OS user"),
+                "Fleetd.main must report the member trust model");
+        assertTrue(contains(appender, "memberCredentials: absent or empty"),
+                "Fleetd.main must report an absent memberCredentials policy");
+        assertTrue(contains(appender, "exhaustedPattern: profile(s) [worker] have no exhaustedPattern configured"),
+                "Fleetd.main must report profiles without exhaustedPattern");
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdUsageLimitFixWarningTest.java

@@ -0,0 +1,75 @@
+package dev.ltms.fleet;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #446 follow-up (criterion 2): {@code exhaustionSink} used to build the "name the fix"
+ * WARNING inline with two SLF4J {@code {}}-placeholder {@code log.warn} calls. Nothing pinned
+ * that text — a mutation battery run against the merged PR #457 proved it: renaming the leading
+ * {@code "usage-limit fix:"} tag to {@code "MUTANT no fix named:"} at both call sites left all
+ * 1586 existing tests green. This class is what makes that mutation red.
+ *
+ * <p>{@link Fleetd#usageLimitFixWarning} and {@link Fleetd#usageLimitFixWarningNoModel} are the
+ * extracted call sites {@code exhaustionSink} actually invokes — {@code log.warn} is called with
+ * each method's return value as a single, already-formatted argument, so the string these tests
+ * assert on is byte-identical to what {@code fleetd.out} receives. Same extracted-static-method +
+ * dedicated-test idiom as {@link Fleetd#exhaustedPatternCoverageLine} /
+ * {@link FleetdPatternCoverageLineTest}, chosen over the {@link ch.qos.logback.core.read.ListAppender}
+ * idiom {@code ExhaustedPatternGapReportTest} uses — this warning is built once, at one call site,
+ * from a single method with no branching inside the message itself, so there is no real log
+ * plumbing left to prove; capturing an appender would only add setup/teardown around the same
+ * assertion.
+ *
+ * <p>Each assertion below checks for a specific substring — the leading {@code "usage-limit fix:"}
+ * tag an operator would grep the log for, the profile name, the model name, and the literal
+ * action ({@code enabled: false} under {@code models.allow}) — rather than the whole sentence.
+ * Criterion 2 promised those facts, not a particular wording of the prose that carries them; a
+ * whole-sentence {@code assertEquals} breaks on every copy-edit of that surrounding prose and
+ * teaches the next person to delete the test rather than read why it failed. The leading tag is
+ * pinned on purpose, unlike the rest of the sentence: it is the identifying label the fleetd #446
+ * follow-up's own mutation battery renamed ({@code "usage-limit fix:"} → {@code "MUTANT no fix
+ * named:"}) to prove this class did not yet catch a tag rename — only the three-fact substrings
+ * above would have missed it, since none of them mention the tag itself.
+ */
+class FleetdUsageLimitFixWarningTest {
+
+    @Test
+    @DisplayName("usageLimitFixWarning names the profile, the model, and the models.allow fix")
+    void usageLimitFixWarningNamesProfileModelAndFix() {
+        String warning = Fleetd.usageLimitFixWarning("terra", "claude-opus-5");
+
+        assertTrue(warning.startsWith("usage-limit fix:"),
+                "must carry the grep-able identifying tag: " + warning);
+        assertTrue(warning.contains("terra"), "must name the profile: " + warning);
+        assertTrue(warning.contains("claude-opus-5"), "must name the model: " + warning);
+        assertTrue(warning.contains("enabled: false"), "must name the action: " + warning);
+        assertTrue(warning.contains("models.allow"), "must name where the action goes: " + warning);
+    }
+
+    @Test
+    @DisplayName("usageLimitFixWarning does not cross-name a different profile or model")
+    void usageLimitFixWarningDoesNotCrossName() {
+        String warning = Fleetd.usageLimitFixWarning("terra", "claude-opus-5");
+
+        // Guards against the mutation of swapping the two profile/model arguments at the call
+        // site — a test that only checks "some name appears" cannot catch that.
+        assertTrue(!warning.contains("sonnet"), "must not name an unrelated model: " + warning);
+    }
+
+    @Test
+    @DisplayName("usageLimitFixWarningNoModel names the profile but no fix, since none exists")
+    void usageLimitFixWarningNoModelNamesProfileNotAFix() {
+        String warning = Fleetd.usageLimitFixWarningNoModel("gx", 1800);
+
+        assertTrue(warning.startsWith("usage-limit fix:"),
+                "must carry the grep-able identifying tag: " + warning);
+        assertTrue(warning.contains("gx"), "must name the profile: " + warning);
+        assertTrue(warning.contains("1800"), "must name the quarantine duration it falls back to: " + warning);
+        assertTrue(!warning.contains("enabled: false"),
+                "a profile with no model: configured has no models.allow entry to flip — the "
+                        + "fallback message must not claim one exists: " + warning);
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/config/ConfigRefProfileCoverageTest.java

@@ -161,7 +161,15 @@ class ConfigRefProfileCoverageTest {
         // the #323 bug. Only ideProjectDir and worktreeGroup were saved by a behavioural test in
         // ConfigRefTest; the other two had none. So: growing this set now requires editing this
         // line as well, which is a visible, deliberate diff rather than a quiet one.
-        assertEquals(Set.of("weight", "maxLoad", "credentialId"), excluded,
+        //
+        // fleetd #446: "exhaustedPattern" joined the set legitimately — it moved from compiled
+        // once at Fleetd.main startup to read live, cached by profile name, through
+        // LiveExhaustedPatterns (see that class's doc and ConfigRef's class doc). Unlike the #323
+        // hazard this comment warns about, it is pinned by its OWN behavioural tests:
+        // FleetdExhaustionDetectionArmedWiringTest proves the reload takes effect with no restart,
+        // and the same test rerun against the pre-#446 compile site (see that ticket's report) is
+        // what proves the assertion would have failed before the fix.
+        assertEquals(Set.of("weight", "maxLoad", "credentialId", "exhaustedPattern"), excluded,
                 "ConfigRef.LAUNCH_SETTINGS_EXCLUDED changed. A component belongs in it ONLY if it "
                         + "is read live off the config supplier, not baked into a launcher at "
                         + "startup. If you are adding one to silence this test, that is fleetd #323 "

fleetd/src/test/java/dev/ltms/fleet/config/ConfigRefTest.java

@@ -339,12 +339,18 @@ class ConfigRefTest {
     }
 
     /**
-     * CB-578 stage B: exhaustedPattern is compiled once into Fleetd.main's pattern map at startup
-     * (see ExhaustedPatternLookup), so a reload never re-reads it — a changed pattern must be
-     * reported deferred exactly like model/baseUrl, not silently claimed as applied.
+     * fleetd #446: exhaustedPattern moved from deferred to hot — {@code LiveExhaustedPatterns}
+     * reads {@code config.get()} fresh on every lookup, cached by profile name (not baked into a
+     * frozen map inside {@code Fleetd.main} any more, the way it was before this ticket, and the
+     * way its sibling {@code errorPattern} still is — see
+     * {@code changingAProfilesErrorPatternIsReportedAsDeferred} right below for the contrast).
+     * So a reload that changes ONLY exhaustedPattern must report a plain "config reloaded", with
+     * nothing deferred — the opposite of what this test asserted before fleetd #446, when
+     * ConfigRef.sameLaunchSettings still compared exhaustedPattern and this reload reported it as
+     * one of the changes needing a restart.
      */
     @Test
-    void changingAProfilesExhaustedPatternIsReportedAsDeferred(@TempDir Path dir) throws Exception {
+    void changingAProfilesExhaustedPatternTakesEffectWithNoRestart(@TempDir Path dir) throws Exception {
         Path f = dir.resolve("fleetd.yaml");
         Files.writeString(f, """
                 bind:
@@ -379,17 +385,21 @@ class ConfigRefTest {
         ConfigRef.Outcome out = ref.reload();
 
         assertTrue(out.applied());
-        assertEquals(1, out.deferred().size(), out.deferred().toString());
-        assertTrue(out.deferred().getFirst().contains("sonnet"), out.deferred().toString());
-        assertTrue(out.deferred().getFirst().contains("launch settings"), out.deferred().toString());
-        // The snapshot still carries the new value — a restart is what makes it take effect.
+        assertEquals(0, out.deferred().size(),
+                "exhaustedPattern is hot since fleetd #446 — changing only this key must not be "
+                        + "reported as needing a restart: " + out.deferred());
+        assertEquals(0, out.split().size(), out.split().toString());
+        assertEquals("config reloaded", out.summary());
+        // The snapshot carries the new value immediately — this IS the live value now, not
+        // merely what a restart would eventually pick up.
         assertEquals("rate limit exceeded", ref.get().profiles().get("sonnet").exhaustedPattern());
     }
 
     /**
      * fleetd #201 Unit 5: errorPattern is compiled once into Fleetd.main's backend-error pattern
-     * map at startup (see BackendErrorPatternLookup), the same way exhaustedPattern is above — a
-     * reload never re-reads it either, so a changed value must be reported deferred.
+     * map at startup (see BackendErrorPatternLookup) — unlike its sibling exhaustedPattern above,
+     * which fleetd #446 made hot, errorPattern was scoped OUT of that ticket on purpose and stays
+     * deferred: a reload still never re-reads it, so a changed value must be reported deferred.
      */
     @Test
     void changingAProfilesErrorPatternIsReportedAsDeferred(@TempDir Path dir) throws Exception {

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

@@ -17,15 +17,83 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
  * SHELL directly (never {@code claude}, so no subscription/token involvement) and always tears
  * the throwaway space down.
  *
+ * <p>The seed shell's own startup (restoring its session, printing its banner) is asynchronous
+ * and its length is not a fleetd contract — measured here at ~2.5s on one host (fleetd #449).
+ * The old version used two fixed sleeps: 1000ms before typing, then 800ms before reading. It
+ * failed, and the pane showed the typed line followed by the startup banner with no command
+ * output at all — which looks, at a glance, exactly like the env map never reaching the shell.
+ * So this polls for a real signal instead of guessing a sleep length.
+ *
+ * <p><strong>What was measured, and what was not.</strong> Polling fixes it: 5 standalone runs
+ * green. The load-bearing half is {@link #waitForText}. With {@link #SHELL_READY_TIMEOUT_MS}
+ * set to 0 — so input is typed at once, with no settle wait at all — the test still passed 3 of
+ * 3. So the proven cause is the 800ms READ deadline being too short, not the 1000ms write delay.
+ * Note the direction, because it matters: typing at 0ms works where typing at 1000ms failed. The
+ * earlier explanation for this test — that input typed before the prompt is swallowed by the
+ * shell's startup — is therefore NOT supported by any measurement here. Please do not repeat it
+ * as the reason; if it were true, 0ms would be worse than 1000ms, and it is better.
+ *
+ * <p>{@link #waitUntilSettled} is kept as cheap insurance against that swallow case, not because
+ * anyone showed it was needed. If you want to delete it, the honest test is whether you can make
+ * this test fail by typing early. Nobody has managed that yet.
+ *
  * <p>Tagged {@code contract}; run with {@code mvn test -Pcontract}.
  */
 @Tag("contract")
 class AgentControlContractTest {
 
+    private static final long POLL_INTERVAL_MS = 150;
+    /** Bound for the seed shell to settle: observed ~2.5s three times running; this leaves headroom. */
+    private static final long SHELL_READY_TIMEOUT_MS = 8_000;
+    /** Bound for the typed command's output to appear once the shell is ready: observed ~0.2s. */
+    private static final long OUTPUT_TIMEOUT_MS = 5_000;
+
     private boolean noSocket() {
         return !Files.exists(UnixSocketHerdrClient.defaultSocketPath());
     }
 
+    private static String readPane(UnixSocketHerdrClient herdr, String paneId) {
+        return herdr.call("pane.read", Map.of("pane_id", paneId, "source", "visible"))
+                .path("read").path("text").asText("");
+    }
+
+    /**
+     * Poll {@code pane.read} until two consecutive reads come back identical — the shell's own
+     * startup output (restore banner, prompt) has stopped changing — or {@code timeoutMs} elapses.
+     * Never asserts by itself; the caller's own assertion is what actually verifies the outcome.
+     * Setting {@code timeoutMs} to 0 skips the wait entirely and the test still passes here, so
+     * treat this as insurance rather than as the fix — see the class javadoc.
+     */
+    private static String waitUntilSettled(UnixSocketHerdrClient herdr, String paneId, long timeoutMs)
+            throws InterruptedException {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        String previous = null;
+        while (System.currentTimeMillis() < deadline) {
+            Thread.sleep(POLL_INTERVAL_MS);
+            String current = readPane(herdr, paneId);
+            if (current.equals(previous) && !current.isBlank()) {
+                return current;
+            }
+            previous = current;
+        }
+        return previous == null ? "" : previous;
+    }
+
+    /** Poll {@code pane.read} until {@code needle} appears or {@code timeoutMs} elapses. */
+    private static String waitForText(UnixSocketHerdrClient herdr, String paneId, String needle, long timeoutMs)
+            throws InterruptedException {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        String last = "";
+        while (System.currentTimeMillis() < deadline) {
+            last = readPane(herdr, paneId);
+            if (last.contains(needle)) {
+                return last;
+            }
+            Thread.sleep(POLL_INTERVAL_MS);
+        }
+        return last;
+    }
+
     @Test
     void tabCreateInjectsEnvIntoTheSeedShell() throws Exception {
         assumeTrue(!noSocket(), "no herdr socket — skipping");
@@ -36,15 +104,13 @@ class AgentControlContractTest {
                     Map.of("ANTHROPIC_BASE_URL", "http://gx00.gw:8000"));
             try {
                 assertNotNull(tab.rootPaneId(), "tab.create must return the seed pane");
-                Thread.sleep(1000); // let the seed shell reach its prompt
+                waitUntilSettled(herdr, tab.rootPaneId(), SHELL_READY_TIMEOUT_MS);
                 herdr.call("pane.send_input", Map.of(
                         "pane_id", tab.rootPaneId(),
                         "text", "printf 'PROBE_BASE=[%s]\\n' \"$ANTHROPIC_BASE_URL\"",
                         "keys", List.of("enter")));
-                Thread.sleep(800);
-                String visible = herdr.call("pane.read",
-                                Map.of("pane_id", tab.rootPaneId(), "source", "visible"))
-                        .path("read").path("text").asText("");
+                String visible = waitForText(herdr, tab.rootPaneId(),
+                        "PROBE_BASE=[http://gx00.gw:8000]", OUTPUT_TIMEOUT_MS);
                 assertTrue(visible.contains("PROBE_BASE=[http://gx00.gw:8000]"),
                         "env map must reach the seed shell; saw: " + visible);
             } finally {

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

@@ -14,7 +14,7 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
  * Contract test against a REAL running herdr. Tagged {@code contract} so it is
  * excluded from {@code mvn test}; run it with {@code mvn test -Pcontract}. It fails
  * loudly if herdr drifts from the protocol {@code fleetd} was built against
- * (0.7.0, protocol 14) — catching breakage that unit tests with canned frames cannot.
+ * (0.8.0, protocol 19) — catching breakage that unit tests with canned frames cannot.
  */
 @Tag("contract")
 class HerdrContractTest {
@@ -24,13 +24,13 @@ class HerdrContractTest {
     }
 
     @Test
-    void pingReturnsProtocol14() {
+    void pingReturnsProtocol19() {
         assumeTrue(Files.exists(socket()), "no herdr socket at " + socket() + " — skipping");
         try (UnixSocketHerdrClient herdr = UnixSocketHerdrClient.connect()) {
             JsonNode pong = herdr.call("ping");
             assertEquals("pong", pong.get("type").asText());
-            assertEquals(14, pong.get("protocol").asInt(),
-                    "fleetd is built against herdr protocol 14");
+            assertEquals(19, pong.get("protocol").asInt(),
+                    "fleetd is built against herdr protocol 19");
             assertFalse(pong.get("version").asText().isBlank());
         }
     }

fleetd/src/test/java/dev/ltms/fleet/inject/LiveExhaustedPatternsTest.java

@@ -0,0 +1,138 @@
+package dev.ltms.fleet.inject;
+
+import dev.ltms.fleet.config.FleetConfig;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #446: {@code LiveExhaustedPatterns} is the mechanism that makes {@code exhaustedPattern}
+ * hot — read fresh off a config supplier per lookup, cached by profile name. This is the direct,
+ * minimal unit test of that mechanism: read the pattern once, mutate the backing "config", read
+ * again, and assert the second read saw the new value — the exact hotness proof shape.
+ *
+ * <p>This class alone does not prove {@code Fleetd.main} actually wires the live source into
+ * production — {@link dev.ltms.fleet.FleetdExhaustionDetectionArmedWiringTest} and {@code
+ * CompletionResolverTest}'s {@code ExhaustedPatternLookup} tests do that at the wiring level. What
+ * this class proves is that the mechanism itself is genuinely live and genuinely cached, not that
+ * it is used.
+ */
+class LiveExhaustedPatternsTest {
+
+    private static FleetConfig.Profile profileWithPattern(String pattern) {
+        return new FleetConfig.Profile("terra", "http://gx00.gw:8000", "terra",
+                null, null, null, "tab", "fleetd-workers", "w #{n}", null, null, null,
+                null, null, null, null, null, null, null, pattern);
+    }
+
+    /** Simple mutable holder standing in for {@code ConfigRef} — swapped, never mutated in place. */
+    private static final class MutableProfiles {
+        private volatile Map<String, FleetConfig.Profile> profiles;
+
+        MutableProfiles(Map<String, FleetConfig.Profile> initial) {
+            this.profiles = initial;
+        }
+
+        Map<String, FleetConfig.Profile> get() {
+            return profiles;
+        }
+
+        void set(Map<String, FleetConfig.Profile> fresh) {
+            this.profiles = fresh;
+        }
+    }
+
+    @Test
+    @DisplayName("patternFor reads the live config: read, change, read again, see the change — no restart")
+    void patternForIsHotAcrossAConfigChange() {
+        MutableProfiles live = new MutableProfiles(Map.of("terra", profileWithPattern("usage limit")));
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(live::get);
+
+        Pattern before = patterns.patternFor("terra");
+        assertTrue(before.matcher("your usage limit has been reached").find());
+
+        // Change the "config" — the exact thing a reload does to ConfigRef in production.
+        live.set(Map.of("terra", profileWithPattern("rate limit exceeded")));
+
+        Pattern after = patterns.patternFor("terra");
+        assertTrue(after.matcher("429: rate limit exceeded, try later").find(),
+                "the second read must see the NEW pattern text with no restart");
+        assertFalse(after.matcher("your usage limit has been reached").find(),
+                "the second read must have actually recompiled — not just reused a stale match");
+    }
+
+    @Test
+    @DisplayName("armed follows the same live read: on, then off, after a config change — no restart")
+    void armedIsHotAcrossAConfigChange() {
+        MutableProfiles live = new MutableProfiles(Map.of("terra", profileWithPattern("usage limit")));
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(live::get);
+
+        assertTrue(patterns.armed("terra"));
+
+        live.set(Map.of("terra", profileWithPattern(null)));
+
+        assertFalse(patterns.armed("terra"),
+                "removing exhaustedPattern and 'reloading' must disarm detection with no restart");
+    }
+
+    @Test
+    @DisplayName("an unchanged pattern string is not recompiled — the cache is reused")
+    void unchangedPatternTextReusesTheCompiledInstance() {
+        MutableProfiles live = new MutableProfiles(Map.of("terra", profileWithPattern("usage limit")));
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(live::get);
+
+        Pattern first = patterns.patternFor("terra");
+        // A NEW Profile object, same pattern text — a reload of an unrelated key produces a fresh
+        // FleetConfig even though this profile's own text did not change.
+        live.set(Map.of("terra", profileWithPattern("usage limit")));
+        Pattern second = patterns.patternFor("terra");
+
+        assertSame(first, second,
+                "identical pattern text must reuse the cached compiled Pattern, not recompile it — "
+                        + "recompiling a regex on every check is exactly the cost the cache exists to avoid");
+    }
+
+    @Test
+    @DisplayName("a changed pattern string is recompiled, not silently reused from the cache")
+    void changedPatternTextIsRecompiled() {
+        MutableProfiles live = new MutableProfiles(Map.of("terra", profileWithPattern("usage limit")));
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(live::get);
+
+        Pattern first = patterns.patternFor("terra");
+        live.set(Map.of("terra", profileWithPattern("rate limit")));
+        Pattern second = patterns.patternFor("terra");
+
+        assertNotSame(first, second, "changed pattern text must produce a freshly compiled Pattern");
+    }
+
+    @Test
+    void unknownProfileReturnsNullAndUnarmed() {
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(Map::of);
+        assertNull(patterns.patternFor("nope"));
+        assertFalse(patterns.armed("nope"));
+    }
+
+    @Test
+    void nullProfileNameReturnsNullAndUnarmed() {
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(
+                () -> Map.of("terra", profileWithPattern("usage limit")));
+        assertNull(patterns.patternFor(null));
+        assertFalse(patterns.armed(null));
+    }
+
+    @Test
+    void profileWithNoPatternConfiguredReturnsNull() {
+        LiveExhaustedPatterns patterns = new LiveExhaustedPatterns(
+                () -> Map.of("terra", profileWithPattern(null)));
+        assertNull(patterns.patternFor("terra"));
+        assertFalse(patterns.armed("terra"));
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetMcpTest.java

@@ -352,7 +352,7 @@ class FleetMcpTest {
                 fail("a lead fleet_reply must not publish to the worker inbox");
             }
             @Override public List<InboxMessage> peek(String target) { return List.of(); }
-            @Override public void ack(String target, String msgId) { }
+            @Override public boolean ack(String target, String msgId) { return false; }
         };
         MessageService leadMessages = new MessageService(agents, new Injector(agents), new Rendezvous(),
                 inboxThatRejectsPublishes);
@@ -1386,6 +1386,10 @@ class FleetMcpTest {
 
     @Test
     void bridgeAckReturnsConfirmationForValidArgs() {
+        // fleet_ack only reports success for a msgId actually queued in the target's inbox
+        // (fleetd #437) — publish one via the inbox directly rather than asserting on a
+        // fabricated id nothing ever queued.
+        inbox.publish("term_a", "msg-1", "queued reply");
         McpSchema.CallToolResult res = FleetMcp.ack(messages, "term_a", "msg-1");
         assertNotEquals(Boolean.TRUE, res.isError());
         assertTrue(textOf(res).contains("msg-1"), "response should mention the msgId");
@@ -1398,6 +1402,26 @@ class FleetMcpTest {
         assertTrue(FleetMcp.ack(messages, "  ", "msg-1").isError());
     }
 
+    @Test
+    void bridgeAckOfAnIdInNoInboxIsAnError() {
+        // fleetd #437: fleet_ack used to say "acknowledged <msgId>" for a message it never
+        // touched, because nothing in the chain reported hit vs. miss. "never-queued" is in no
+        // inbox at all, so this must error rather than claim success.
+        McpSchema.CallToolResult res = FleetMcp.ack(messages, "term_a", "never-queued");
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("never-queued"), textOf(res));
+    }
+
+    @Test
+    void bridgeAckOfACoordIdTargetIsAnErrorNamingFleetPoll() {
+        // A coord-id names a peer lead's held mailbox (LeadChannel/LeadMailbox), never a
+        // worker's ReplyInbox — fleet_ack has no route to it and must say so, pointing at
+        // fleet_poll{coordId} instead of reporting a false "acknowledged".
+        McpSchema.CallToolResult res = FleetMcp.ack(messages, "coord-some-peer", "msg-1");
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("fleet_poll{coordId}"), textOf(res));
+    }
+
     @Test
     void bridgeAckRemovesSpecificReply() {
         // Queue a reply and capture its msgId.
@@ -1411,8 +1435,18 @@ class FleetMcpTest {
         var peeked = messages.drainReplies("term_a");
         assertEquals(1, peeked.size(), "one fresh reply in the inbox");
 
-        // ackReply works (no-op since published with a different UUID, but callable).
-        assertDoesNotThrow(() -> messages.ackReply("term_a", msgId));
+        // fleetd #437: msgId was already drained above (a fresh UUID each publish), so it is no
+        // longer in the inbox — ackReply must now report that miss instead of pretending to ack.
+        assertFalse(messages.ackReply("term_a", msgId));
+    }
+
+    @Test
+    void bridgeAckRemovingARealQueuedReplyReportsSuccessAndRemovesIt() {
+        // The worker path must not change behaviour: acking a reply that IS still in the inbox
+        // still succeeds and still removes it (fleetd #437).
+        inbox.publish("term_a", "real-1", "still queued");
+        assertTrue(messages.ackReply("term_a", "real-1"), "ack of a real queued reply must report true");
+        assertTrue(inbox.peek("term_a").isEmpty(), "the acked reply must be gone from the inbox");
     }
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetProfilesLiveDefaultTest.java

@@ -0,0 +1,114 @@
+package dev.ltms.fleet.mcp;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.WorkspaceControl;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.member.CompositePeerLauncher;
+import dev.ltms.fleet.peer.MemberRole;
+import dev.ltms.fleet.peer.PeerLauncher;
+import dev.ltms.fleet.peer.SpawnRequest;
+import dev.ltms.fleet.placement.BackendQuarantine;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #425: {@code fleet_profiles}' {@code "default"} field was captured once at boot
+ * ({@code cfg.effectiveDefaultProfile()}, frozen into {@code CompositePeerLauncher.defaultProfile}
+ * at construction) while an unqualified spawn resolves the same underlying key
+ * ({@code fleet.developers}' first entry) live, on every call. Reordering {@code fleet.developers}
+ * and reloading changed where a spawn landed without ever changing what {@code fleet_profiles}
+ * reported — a lead following {@code CLAUDE.md}'s "check {@code fleet_profiles} once per session"
+ * instruction was told a stale answer.
+ *
+ * <p>This test drives the exact caller {@code fleet_profiles} uses —
+ * {@link FleetMcp#profilesView(PeerLauncher, FleetMcp.QuarantineSource, FleetMcp.OutageSource)} —
+ * against a real, reloadable {@link ConfigRef}, so it fails if the reporting path is ever recoupled
+ * to a frozen value instead of {@link CompositePeerLauncher#defaultProfile()}'s live answer.
+ */
+class FleetProfilesLiveDefaultTest {
+
+    /** A minimal fleetd.yaml whose dev pool is {@code profilesInOrder}, in that definition order. */
+    private static String yamlWithDevPool(String... profilesInOrder) {
+        StringBuilder devPool = new StringBuilder();
+        for (int i = 0; i < profilesInOrder.length; i++) {
+            devPool.append("    slot").append(i).append(":\n      profile: ")
+                    .append(profilesInOrder[i]).append('\n');
+        }
+        return """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: ~/.config/herdr/herdr.sock
+                profiles:
+                  opus:
+                    baseUrl: http://gx00.gw:8000
+                    model: opus-coder
+                  sonnet:
+                    baseUrl: http://gx00.gw:8000
+                    model: sonnet-coder
+                guard:
+                  offSubscriptionHosts:
+                    - gx00.gw
+                fleet:
+                  developers:
+                """ + devPool;
+    }
+
+    @Test
+    void fleetProfilesDefaultTracksALiveDevPoolReorderAfterReload(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, yamlWithDevPool("opus", "sonnet"));
+        ConfigRef ref = new ConfigRef(f, FleetConfig.load(f));
+
+        Map<String, FleetConfig.Profile> profiles = Map.of(
+                "opus", new FleetConfig.Profile("opus", "http://gx00.gw:8000", "opus-coder", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, null, null),
+                "sonnet", new FleetConfig.Profile("sonnet", "http://gx00.gw:8000", "sonnet-coder", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, null, null));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "opus", _ -> null);
+        PeerLauncher workers = new CompositePeerLauncher(
+                List.of(adapter), "opus", ref, _ -> 0, BackendQuarantine.none());
+
+        assertReportedDefaultMatchesAnUnqualifiedSpawn(workers, "opus");
+
+        Files.writeString(f, yamlWithDevPool("sonnet", "opus"));
+        ConfigRef.Outcome out = ref.reload();
+        assertTrue(out.applied(), () -> "reload should apply cleanly: " + out.error());
+
+        assertReportedDefaultMatchesAnUnqualifiedSpawn(workers, "sonnet");
+    }
+
+    /**
+     * Asserts BOTH that {@code fleet_profiles}' {@code "default"} equals {@code expected}, AND that
+     * it equals what a real unqualified {@code MemberRole#DEV} spawn actually gets placed on right
+     * now — the two facts fleetd #425 found disagreeing.
+     */
+    private static void assertReportedDefaultMatchesAnUnqualifiedSpawn(PeerLauncher workers, String expected) {
+        Map<String, Object> view = FleetMcp.profilesView(
+                workers, FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
+        assertEquals(expected, view.get("default"),
+                "fleet_profiles' \"default\" must be the live dev-pool answer, not a boot-time snapshot");
+
+        String placed = workers.spawn(
+                new SpawnRequest(null, null, null, null, null, MemberRole.DEV)).profile();
+        assertEquals(expected, placed,
+                "sanity: the profile an unqualified dev spawn actually lands on");
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetProfilesQuarantineModelReasonFieldsTest.java

@@ -0,0 +1,111 @@
+package dev.ltms.fleet.mcp;
+
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.WorkspaceControl;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.peer.PeerLauncher;
+import dev.ltms.fleet.placement.BackendQuarantine;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #446 follow-up (criterion 3): a mutation battery run against the merged PR #457 proved
+ * that deleting either {@code row.put("model", model);} or {@code row.put("reason", reason);}
+ * from {@link FleetMcp#profilesView} left all 1586 existing tests green — nothing exercised a
+ * {@link FleetMcp.QuarantineSource} whose {@code modelFor}/{@code reasonFor} actually return a
+ * value. This class is what makes both deletions red, and also pins the two "must be omitted, not
+ * emitted as null/blank" branches those two {@code if} guards exist for — {@link
+ * FleetMcpTest#profilesReportsAQuarantinedCredential} covers the credential/quarantine-duration
+ * shape but supplies neither function, so it cannot distinguish "the field is missing" from "the
+ * field was never asked for".
+ */
+class FleetProfilesQuarantineModelReasonFieldsTest {
+
+    private static final String PROFILE = "terra";
+    private static final String CREDENTIAL = "cred-terra";
+
+    private static PeerLauncher launcher(FakeHerdr h) {
+        FleetConfig.Profile profile = new FleetConfig.Profile(
+                PROFILE, "http://gx00.gw:8000", "claude-opus-5", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        return new ClaudeCodeLauncher(new AgentControl(h), new WorkspaceControl(h),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(PROFILE, profile), PROFILE, _ -> "tok");
+    }
+
+    private static BackendQuarantine quarantined(String credentialId) {
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine(credentialId);
+        return quarantine;
+    }
+
+    private static String textOf(McpSchema.CallToolResult r) {
+        return ((McpSchema.TextContent) r.content().getFirst()).text();
+    }
+
+    @Test
+    void quarantinedRowNamesTheModelAndTheReason() {
+        FakeHerdr h = new FakeHerdr();
+        FleetMcp.QuarantineSource source = new FleetMcp.QuarantineSource(
+                _ -> CREDENTIAL, quarantined(CREDENTIAL), _ -> true,
+                _ -> "claude-opus-5", _ -> "The usage limit has been reached");
+
+        McpSchema.CallToolResult res = FleetMcp.profiles(launcher(h), source);
+        assertNotEquals(Boolean.TRUE, res.isError());
+        String out = textOf(res);
+
+        assertTrue(out.contains("\"quarantined\""), out);
+        assertTrue(out.contains("\"model\":\"claude-opus-5\""),
+                "the quarantined row must name the model the fix applies to: " + out);
+        assertTrue(out.contains("\"reason\":\"The usage limit has been reached\""),
+                "the quarantined row must name why it was quarantined: " + out);
+    }
+
+    @Test
+    void quarantinedRowOmitsModelWhenNoneIsConfigured() {
+        FakeHerdr h = new FakeHerdr();
+        // null and "" (blank) both exercise the same production guard (`!model.isBlank()`) —
+        // covered together since both must produce the identical outcome: the key absent.
+        for (String noModel : new String[] {null, "", "   "}) {
+            FleetMcp.QuarantineSource source = new FleetMcp.QuarantineSource(
+                    _ -> CREDENTIAL, quarantined(CREDENTIAL), _ -> true,
+                    _ -> noModel, _ -> "some reason");
+
+            McpSchema.CallToolResult res = FleetMcp.profiles(launcher(h), source);
+            String out = textOf(res);
+
+            assertTrue(out.contains("\"quarantined\""), out);
+            assertFalse(out.contains("\"model\""),
+                    "modelFor returned " + (noModel == null ? "null" : "'" + noModel + "'")
+                            + " — the key must be ABSENT, not emitted as null or an empty string: " + out);
+        }
+    }
+
+    @Test
+    void quarantinedRowOmitsReasonWhenNoneIsKnown() {
+        FakeHerdr h = new FakeHerdr();
+        for (String noReason : new String[] {null, "", "  "}) {
+            FleetMcp.QuarantineSource source = new FleetMcp.QuarantineSource(
+                    _ -> CREDENTIAL, quarantined(CREDENTIAL), _ -> true,
+                    _ -> "claude-opus-5", _ -> noReason);
+
+            McpSchema.CallToolResult res = FleetMcp.profiles(launcher(h), source);
+            String out = textOf(res);
+
+            assertTrue(out.contains("\"quarantined\""), out);
+            assertFalse(out.contains("\"reason\""),
+                    "reasonFor returned " + (noReason == null ? "null" : "'" + noReason + "'")
+                            + " — the key must be ABSENT, not emitted as null or an empty string: " + out);
+        }
+    }
+}

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

@@ -20,6 +20,7 @@ import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.peer.SpawnRequest;
 import dev.ltms.fleet.placement.BackendOutagePolicy;
 import dev.ltms.fleet.placement.BackendQuarantine;
+import dev.ltms.fleet.placement.PlacementDecision;
 import dev.ltms.fleet.placement.PlacementException;
 import dev.ltms.fleet.placement.PlacementPolicies;
 import org.junit.jupiter.api.Test;
@@ -944,6 +945,89 @@ class CompositePeerLauncherTest {
         assertTrue(e.getMessage().contains("maxLoad"), e.getMessage());
     }
 
+    // ── fleetd #425: defaultProfile()/defaultProfileFor() must track a live reload ─────────────
+
+    /** A minimal fleetd.yaml whose dev pool is {@code profilesInOrder}, in that definition order. */
+    private static String yamlWithDevPool(String... profilesInOrder) {
+        StringBuilder devPool = new StringBuilder();
+        for (int i = 0; i < profilesInOrder.length; i++) {
+            devPool.append("    slot").append(i).append(":\n      profile: ")
+                    .append(profilesInOrder[i]).append('\n');
+        }
+        return """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: ~/.config/herdr/herdr.sock
+                profiles:
+                  opus:
+                    baseUrl: http://gx00.gw:8000
+                    model: opus-coder
+                  sonnet:
+                    baseUrl: http://gx00.gw:8000
+                    model: sonnet-coder
+                guard:
+                  offSubscriptionHosts:
+                    - gx00.gw
+                fleet:
+                  developers:
+                """ + devPool;
+    }
+
+    /**
+     * Criterion 1 (fleetd #425): reorder {@code fleet.developers}, reload, and assert the reported
+     * default ({@link CompositePeerLauncher#defaultProfile()} — what {@code fleet_profiles}' {@code
+     * "default"} is built from, see {@code FleetMcp.profilesView}) matches what an unqualified
+     * {@code MemberRole#DEV} spawn is actually placed on, both before and after the reorder. Asserts
+     * {@code applied()} so the test proves the reload actually took, not that nothing changed.
+     */
+    @Test
+    void defaultProfileTracksALiveDevPoolReorderAfterReload(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, yamlWithDevPool("opus", "sonnet"));
+        ConfigRef ref = new ConfigRef(f, FleetConfig.load(f));
+
+        FakeHerdr herdr = new FakeHerdr();
+        StubLauncher adapter = new StubLauncher("claude", herdr, threeProfiles(), "opus", Set.of());
+        CompositePeerLauncher composite = new CompositePeerLauncher(
+                List.of(adapter), "opus", ref, _ -> 0, BackendQuarantine.none());
+
+        assertEquals("opus", composite.defaultProfile(),
+                "reported default starts at the dev pool's first entry");
+        assertEquals("opus", composite.spawn(
+                        new SpawnRequest(null, null, null, null, null, MemberRole.DEV)).profile(),
+                "an unqualified dev spawn must land on the same profile that was just reported");
+
+        Files.writeString(f, yamlWithDevPool("sonnet", "opus"));
+        ConfigRef.Outcome out = ref.reload();
+        assertTrue(out.applied(), () -> "reload should apply cleanly: " + out.error());
+
+        assertEquals("sonnet", composite.defaultProfile(),
+                "the reported default must follow the reorder with no daemon restart");
+        assertEquals("sonnet", composite.spawn(
+                        new SpawnRequest(null, null, null, null, null, MemberRole.DEV)).profile(),
+                "and it must still be exactly what an unqualified spawn actually gets");
+    }
+
+    /**
+     * Criterion 2 — the mirror, and the load-bearing half (fleetd #425): with NOTHING configured (no
+     * profiles at all, hence an empty pool for every role), the frozen {@code defaultProfile} field
+     * is still what gets reported. A fix that always returns {@code poolFor(role).getFirst()} with no
+     * empty-pool fallback throws or returns the wrong thing here even though criterion 1 above still
+     * passes — this is the test that catches it.
+     */
+    @Test
+    void defaultProfileFallsBackToTheFrozenFieldWhenNothingIsConfiguredAtAll() {
+        FakeHerdr herdr = new FakeHerdr();
+        StubLauncher adapter = new StubLauncher("claude", herdr, Map.of(), "opus", Set.of());
+        CompositePeerLauncher composite = new CompositePeerLauncher(
+                List.of(adapter), "opus", Map.of(), PlacementPolicies.fixed(), _ -> 0);
+
+        assertEquals("opus", composite.defaultProfile(),
+                "with no profiles configured at all, the frozen field is the only answer available");
+        assertEquals("opus", composite.defaultProfileFor(MemberRole.DEV));
+    }
+
     // ── CB-578 stage B: a BACKEND_EXHAUSTED classification quarantines the credential ──────────
 
     @Test
@@ -1010,6 +1094,98 @@ class CompositePeerLauncherTest {
         assertEquals(0, adapter.spawnCount("sol"));
     }
 
+    /**
+     * fleetd #425 rework, acceptance 1: {@link CompositePeerLauncher#routedProfileFor} must apply
+     * the SAME quarantine filtering {@link CompositePeerLauncher#spawn} does, under {@code fixed()}
+     * — the DEFAULT placement policy, deliberately not {@code weighted()} (which the regressed
+     * round's own tests all used, and which never exercises {@code FixedPlacementPolicy}'s own
+     * inline filter). This is the exact defect: the previous round's {@code defaultProfileFor}
+     * blindly returns the pool's first entry ("sol", quarantined here) with no awareness of
+     * quarantine at all, which is what turned a routine unqualified spawn into a hard throw once
+     * {@code acquireWithWorktree} pre-resolved through it.
+     */
+    @Test
+    void routedProfileForSkipsAQuarantinedPoolFirstProfileUnderFixedPolicy() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, FleetConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "b", stubWorker("b"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "sol", Set.of());
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        assertEquals("b", composite.routedProfileFor(MemberRole.DEV),
+                "sol (the pool's first entry) is quarantined, so the routed answer must be b");
+        assertEquals("sol", composite.defaultProfileFor(MemberRole.DEV),
+                "sanity: defaultProfileFor stays blind to quarantine — that's the gap routedProfileFor closes");
+        assertEquals(0, adapter.spawnCount("sol"), "routedProfileFor never spawns anything");
+        assertEquals(0, adapter.spawnCount("b"), "routedProfileFor never spawns anything");
+    }
+
+    /**
+     * fleetd #444: {@link PlacementDecision} exists to close the window between {@link
+     * CompositePeerLauncher#place} and {@link CompositePeerLauncher#spawn(SpawnRequest,
+     * PlacementDecision)} — the placement state must be free to move in that window without the
+     * held decision being re-checked against the new state. Every quarantine test above resolves
+     * and spawns in one call, so none of them ever open that window; this test is the one that
+     * does: "sol" is placed FIRST, while nothing is quarantined yet, and only THEN is its
+     * credential quarantined, before the held decision is spawned.
+     *
+     * <p>This is the test that tells the real override apart from the alternative body the ticket
+     * measured: routing {@code decision.profile()} straight to its adapter (the real override)
+     * never re-runs {@code enforceNotQuarantined}, so the spawn against the held decision still
+     * succeeds on sol. Re-entering {@code spawn(req.withProfile(decision.profile()))} instead
+     * lands in the explicit-profile branch, which refuses a now-quarantined sol outright — before
+     * this test existed, replacing the real override's body with that re-entering call left the
+     * whole suite green.
+     */
+    @Test
+    void spawnHonorsAPlacementDecisionEvenAfterItsProfileIsQuarantinedInTheWindowAfterPlace() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, FleetConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "b", stubWorker("b"));
+        // The adapter's OWN fallback default is "b", deliberately different from the profile place()
+        // decides ("sol") — see the note below on why this must not be "sol" too.
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "b", Set.of());
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        // 1. Resolve BEFORE anything is quarantined — sol (definition order first, fixed policy) wins.
+        // composite's own defaultProfile ("sol", the constructor arg above) never enters this: the
+        // pool poolFor(DEV) resolves to is never empty here, so place() only ever reads that field as
+        // a fallback for an empty pool, which this test does not exercise.
+        PlacementDecision decision = composite.place(MemberRole.DEV);
+        assertEquals("sol", decision.profile(), "sanity: nothing is quarantined yet, so sol is placed");
+
+        // 2. Move the placement state IN THE WINDOW between place() and spawn() — sol's credential
+        // is now quarantined. A fresh place()/spawn(req) pair would fall through to b instead; the
+        // held decision must not be re-evaluated against this new state at all.
+        quarantine.quarantine("shared-openai");
+
+        // 3. Spawn against the HELD decision, not a fresh resolve.
+        SpawnRequest req = new SpawnRequest(null, null, null, null, null, MemberRole.DEV);
+        PeerHandle handle = composite.spawn(req, decision);
+
+        assertEquals("sol", handle.profile(),
+                "the decision from place() is honored even though sol is now quarantined");
+        // A fixture whose adapter falls back to "sol" too would let an UNSTAMPED request (one
+        // routed but never given req.withProfile("sol")) land on spawnCount("sol") == 1 by
+        // COINCIDENCE, since StubLauncher.spawn falls back to its own defaultProfile whenever
+        // req.profileName() is blank. Giving the adapter "b" as its fallback instead means only an
+        // actually-stamped request can produce this count — an unstamped one would count against
+        // "b" and this assertion would fail.
+        assertEquals(1, adapter.spawnCount("sol"),
+                "the request that reached the delegate actually carried sol as its profile "
+                        + "(the adapter's own fallback default is 'b', so this can't happen by accident)");
+        assertEquals(0, adapter.spawnCount("b"),
+                "b must never be touched — neither as the decision's profile nor as an unstamped "
+                        + "request's accidental fallback");
+    }
+
     @Test
     void aQuarantineLiftsOnTheInjectedClockAndTheProfileBecomesSpawnableAgain() {
         FakeHerdr herdr = new FakeHerdr();
@@ -1403,6 +1579,37 @@ class CompositePeerLauncherTest {
         assertEquals(0, adapter.spawnCount("local"));
     }
 
+    /**
+     * fleetd #425 rework, acceptance 2: same shape as {@link #fixedPlacementSkipsAnOffModelProfileToo}
+     * above, but through {@link CompositePeerLauncher#routedProfileFor} rather than an actual
+     * {@link CompositePeerLauncher#spawn} — the exact call {@code SessionManager.acquireWithWorktree}
+     * makes to pre-resolve a profile for provisioning. This is the fleetd #429 case named in the
+     * ticket: an operator turns a model off, and an unqualified worktree spawn must still route
+     * around it instead of throwing "names model, which the operator has turned off" — the throw
+     * {@link CompositePeerLauncher#enforceModelEnabled} raises only on the EXPLICIT-profile branch,
+     * which is exactly the branch the regressed round accidentally routed every worktree spawn onto.
+     */
+    @Test
+    void routedProfileForSkipsAModelOffPoolFirstProfileUnderFixedPolicy() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, FleetConfig.Profile> profiles = ordered(
+                "local", stubWorkerModel("local", "deepseek-v4-flash"),
+                "sonnet", stubWorkerModel("sonnet", "claude-sonnet-5"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "local", Set.of());
+        FleetConfig.Models models = new FleetConfig.Models(List.of(
+                new FleetConfig.Models.ModelEntry("deepseek-v4-flash", false)));
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "local", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, BackendQuarantine.none(), NO_OUTAGE,
+                () -> models);
+
+        assertEquals("sonnet", composite.routedProfileFor(MemberRole.DEV),
+                "local (the pool's first entry) names an off model, so the routed answer must be sonnet");
+        assertEquals("local", composite.defaultProfileFor(MemberRole.DEV),
+                "sanity: defaultProfileFor stays blind to model-off — that's the gap routedProfileFor closes");
+        assertEquals(0, adapter.spawnCount("local"), "routedProfileFor never spawns anything");
+        assertEquals(0, adapter.spawnCount("sonnet"), "routedProfileFor never spawns anything");
+    }
+
     /**
      * Criterion 4: turning a model off/on is HOT — no restart — proven through a REAL
      * {@code ConfigRef.reload()}, not a hand-rolled supplier swap. Also proves {@code models} is

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

@@ -16,6 +16,7 @@ import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -221,6 +222,31 @@ class AmqpReplyInboxContractTest {
         }
     }
 
+    @Test
+    void ackReportsHitVsMissAgainstARealBroker() throws Exception {
+        // fleetd #437: fleet_ack said "acknowledged <msgId>" for a message it never touched,
+        // because ReplyInbox.ack() (void) could not tell a hit from a miss. Pin the fixed
+        // boolean contract against a real broker — the adapter fleetd actually runs live.
+        String target = "worker-ack-contract-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
+
+            // Never held for this target at all: must report false, not throw.
+            assertFalse(inbox.ack(target, "never-held"),
+                    "acking a msgId never held for an owned target must report false");
+
+            // A real message: first ack removes it and reports true...
+            inbox.publish(target, "m1", "ack me");
+            assertEquals(1, awaitPeek(inbox, target).size(), "the published reply should be held");
+            assertTrue(inbox.ack(target, "m1"), "acking a held reply must report true");
+            assertTrue(inbox.peek(target).isEmpty(), "an acked reply is dropped");
+
+            // ...and the second ack of the SAME msgId has nothing left to remove: false.
+            assertFalse(inbox.ack(target, "m1"),
+                    "acking the same msgId twice must report false the second time");
+        }
+    }
+
     @Test
     void confirmedPublishDeliversNormally() throws Exception {
         String target = "worker-confirm-" + System.nanoTime();

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

@@ -41,20 +41,20 @@ class InMemoryReplyInboxTest {
     @Test
     void ackRemovesTheMessage() {
         inbox.publish("term_a", "m1", "hello");
-        inbox.ack("term_a", "m1");
+        assertTrue(inbox.ack("term_a", "m1"), "fleetd #437: ack of a real entry must report true");
         assertTrue(inbox.peek("term_a").isEmpty(), "after ack, the message is gone");
     }
 
     @Test
     void ackForUnknownMsgIdIsNoOp() {
         inbox.publish("term_a", "m1", "hello");
-        inbox.ack("term_a", "no-such-id"); // no-op
+        assertFalse(inbox.ack("term_a", "no-such-id"), "fleetd #437: a miss must report false"); // no-op
         assertEquals(1, inbox.peek("term_a").size(), "the published message is still there");
     }
 
     @Test
     void ackForUnknownTargetIsNoOp() {
-        inbox.ack("no-such-target", "m1"); // no-op, should not throw
+        assertFalse(inbox.ack("no-such-target", "m1"), "fleetd #437: a miss must report false"); // no-op, should not throw
     }
 
     @Test
@@ -167,7 +167,7 @@ class InMemoryReplyInboxTest {
     @Test
     void peekAndAckAreNoOpsForUnownedTarget() {
         assertTrue(inbox.peek("term_not_owned").isEmpty());
-        inbox.ack("term_not_owned", "m1"); // no-op, should not throw
+        assertFalse(inbox.ack("term_not_owned", "m1"), "fleetd #437: a miss must report false"); // no-op, should not throw
     }
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/session/SessionManagerTest.java

@@ -6,12 +6,14 @@ import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.fleet.auth.MemberRegistry;
 import dev.ltms.fleet.auth.MemberLifecycle;
+import dev.ltms.fleet.config.ConfigRef;
 import dev.ltms.fleet.config.FleetConfig;
 import dev.ltms.fleet.guard.SubscriptionGuard;
 import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.FakeHerdr;
 import dev.ltms.fleet.herdr.WorkspaceControl;
 import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.member.CompositePeerLauncher;
 import dev.ltms.fleet.msg.TestTurnTokens;
 import dev.ltms.fleet.peer.Capability;
 import dev.ltms.fleet.peer.CharterReceipt;
@@ -20,9 +22,16 @@ import dev.ltms.fleet.peer.PeerHandle;
 import dev.ltms.fleet.peer.PeerLauncher;
 import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.peer.SpawnRequest;
+import dev.ltms.fleet.placement.BackendQuarantine;
+import dev.ltms.fleet.placement.PlacementDecision;
+import dev.ltms.fleet.placement.PlacementPolicies;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.LoggerFactory;
 
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -33,6 +42,7 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.LongSupplier;
+import java.util.function.Supplier;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -1027,6 +1037,11 @@ class SessionManagerTest {
             return delegate.spawn(req);
         }
 
+        @Override
+        public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+            return delegate.spawn(req, decision);
+        }
+
         @Override
         public Set<String> profiles() {
             return delegate.profiles();
@@ -1588,6 +1603,11 @@ class SessionManagerTest {
             throw new UnsupportedOperationException("not reachable — the capability check refuses first");
         }
 
+        @Override
+        public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+            throw new UnsupportedOperationException("not reachable — the capability check refuses first");
+        }
+
         @Override
         public Set<String> profiles() {
             return Set.of("stub-profile");
@@ -1652,6 +1672,11 @@ class SessionManagerTest {
             return delegate.spawn(req);
         }
 
+        @Override
+        public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+            return delegate.spawn(req, decision);
+        }
+
         @Override
         public Set<String> profiles() {
             return delegate.profiles();
@@ -1858,6 +1883,11 @@ class SessionManagerTest {
             return handle;
         }
 
+        @Override
+        public PeerHandle spawn(SpawnRequest req, PlacementDecision decision) {
+            return spawn(req.withProfile(decision.profile()));
+        }
+
         @Override
         public Set<String> profiles() {
             return Set.of("lazy");
@@ -1991,4 +2021,296 @@ class SessionManagerTest {
         sessions.rosterResolved();
         assertEquals(2, handle.callCount(), "once resolved, the id must not be looked up again");
     }
+
+    // ── fleetd #425 criterion 3: acquireWithWorktree must provision for the profile it actually
+    // spawns, never a name resolved before a live pool change is accounted for ────────────────────
+
+    /** Two profiles with distinct {@code cwd}/{@code parityOverlay}, and a dev pool of {@code first,second}. */
+    private static String worktreeReorderYaml(String first, String second) {
+        return """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: ~/.config/herdr/herdr.sock
+                profiles:
+                  a:
+                    baseUrl: http://gx00.gw:8000
+                    model: coder-a
+                  b:
+                    baseUrl: http://gx00.gw:8000
+                    model: coder-b
+                guard:
+                  offSubscriptionHosts:
+                    - gx00.gw
+                fleet:
+                  developers:
+                    slot0:
+                      profile: %s
+                    slot1:
+                      profile: %s
+                """.formatted(first, second);
+    }
+
+    @Test
+    void acquireWithWorktreeProvisionsTheOverlayForTheProfileActuallySpawned(
+            @TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, worktreeReorderYaml("a", "b"));
+        ConfigRef ref = new ConfigRef(f, FleetConfig.load(f));
+
+        Map<String, FleetConfig.Profile> profiles = Map.of(
+                "a", new FleetConfig.Profile("a", "http://gx00.gw:8000", "coder-a", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, "/repo/a", List.of("a.mcp.json")),
+                "b", new FleetConfig.Profile("b", "http://gx00.gw:8000", "coder-b", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, "/repo/b", List.of("b.mcp.json")));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "a", _ -> null);
+        PeerLauncher launcher = new CompositePeerLauncher(
+                List.of(adapter), "a", ref, _ -> 0, BackendQuarantine.none());
+
+        // The pool changes AFTER the composite/launcher is built, and BEFORE the unqualified
+        // worktree spawn — exactly the fleetd #425 scenario: the live pool's first entry is "b" by
+        // the time acquireWithWorktree runs, even though nothing here was rebuilt.
+        Files.writeString(f, worktreeReorderYaml("b", "a"));
+        ConfigRef.Outcome out = ref.reload();
+        assertTrue(out.applied(), () -> "reload should apply cleanly: " + out.error());
+
+        FakeWorktrees worktrees = new FakeWorktrees();
+        SessionManager sessions = new SessionManager(launcher, worktrees, () -> 0L);
+
+        MemberSession s = sessions.acquire(null, null, "/caller",
+                null, new WorktreeRequest("fleetd-425", null));
+
+        assertEquals("b", s.profile(),
+                "the live dev pool now starts at b, so the unqualified spawn must land there");
+        FakeWorktrees.OverlayCall overlay = worktrees.lastOverlay();
+        assertNotNull(overlay, "overlayParity must have been called");
+        assertEquals(List.of("b.mcp.json"), overlay.requested(),
+                "the worktree must be provisioned with profile b's overlay — the one actually "
+                        + "spawned — never a's, the pool's stale first entry");
+    }
+
+    /**
+     * The deterministic, mutation-pinning half of criterion 3: {@code launcher.defaultProfile()}
+     * only ever answers for {@link MemberRole#DEV} (see {@link CompositePeerLauncher#defaultProfile()}),
+     * so resolving a worktree spawn's profile through it — instead of through {@link
+     * PeerLauncher#defaultProfileFor(MemberRole)}, resolved against the CALLER's actual role — picks
+     * the wrong pool's answer for any role other than DEV. No reload or race is needed to see it: an
+     * ARCHITECT pool and a DEV pool that simply disagree, held constant, are enough.
+     */
+    @Test
+    void acquireWithWorktreeForANonDevRoleUsesThatRolesPoolNotTheDevPool() {
+        Map<String, FleetConfig.Profile> profiles = Map.of(
+                "a", new FleetConfig.Profile("a", "http://gx00.gw:8000", "coder-a", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, "/repo/a", List.of("a.mcp.json")),
+                "b", new FleetConfig.Profile("b", "http://gx00.gw:8000", "coder-b", null,
+                        "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                        "worker: {profile} #{n}", null, "/repo/b", List.of("b.mcp.json")));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "a", _ -> null);
+        // developers -> a (first/only entry); architects -> b (first/only entry). The two pools
+        // disagree on purpose, so a role-blind resolution (DEV's answer, "a") is visibly wrong for
+        // an ARCHITECT spawn, which must land on "b".
+        FleetConfig.Fleet fleet = new FleetConfig.Fleet(Map.of(),
+                Map.of("s0", new FleetConfig.Slot("b")),
+                Map.of("s0", new FleetConfig.Slot("a")),
+                Map.of(), null);
+        PeerLauncher launcher = new CompositePeerLauncher(List.of(adapter), "a", profiles,
+                PlacementPolicies.fixed(), _ -> 0, fleet);
+
+        FakeWorktrees worktrees = new FakeWorktrees();
+        SessionManager sessions = new SessionManager(launcher, worktrees, () -> 0L);
+
+        MemberSession s = sessions.acquire(null, MemberRole.ARCHITECT, null, "/caller",
+                null, new WorktreeRequest("fleetd-425b", null));
+
+        assertEquals("b", s.profile(),
+                "an unqualified ARCHITECT worktree spawn must land on the architect pool's profile");
+        FakeWorktrees.OverlayCall overlay = worktrees.lastOverlay();
+        assertNotNull(overlay, "overlayParity must have been called");
+        assertEquals(List.of("b.mcp.json"), overlay.requested(),
+                "the worktree must be provisioned with profile b's overlay — the ARCHITECT pool's "
+                        + "answer, the one actually spawned — never a's, the DEV pool's answer that "
+                        + "launcher.defaultProfile() alone would have given");
+    }
+
+    /**
+     * fleetd #425 rework, acceptance 3: repoRoot, parityOverlay, AND the actual spawn must all name
+     * the SAME routed profile, proven on the ROUTED path — a quarantine skips the pool's first entry
+     * — not just the "pool reordered by a live reload" path the two tests above already cover.
+     *
+     * <p>This is the exact regression the rework fixes: the first round resolved
+     * {@code acquireWithWorktree}'s profile through {@code launcher.defaultProfileFor(memberRole)},
+     * which is blind to quarantine and just returns the pool's first entry ("a" here, quarantined).
+     * That name went on to provision repoRoot/overlay for "a", and then the spawn itself — now an
+     * EXPLICIT-profile spawn naming "a" — hit {@code CompositePeerLauncher.enforceNotQuarantined}
+     * and threw, where the pre-fix code (a blank-profile spawn) would have routed around "a" onto
+     * "b" without any trouble. {@code launcher.routedProfileFor(memberRole)} closes that gap by
+     * running the SAME quarantine-aware selection {@code spawn} itself uses, so all three — repoRoot,
+     * overlay, and the spawn — land on "b" together.
+     */
+    @Test
+    void acquireWithWorktreeRoutesAroundAQuarantinedPoolFirstProfile() {
+        Map<String, FleetConfig.Profile> profiles = new LinkedHashMap<>();
+        profiles.put("a", new FleetConfig.Profile("a", "http://gx00.gw:8000", "coder-a", null,
+                "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, "/repo/a", List.of("a.mcp.json"),
+                null, null, null, null, null, null, null, null, "shared-cred", null));
+        profiles.put("b", new FleetConfig.Profile("b", "http://gx00.gw:8000", "coder-b", null,
+                "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, "/repo/b", List.of("b.mcp.json")));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "a", _ -> null);
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-cred");
+        PeerLauncher launcher = new CompositePeerLauncher(List.of(adapter), "a", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        FakeWorktrees worktrees = new FakeWorktrees();
+        SessionManager sessions = new SessionManager(launcher, worktrees, () -> 0L);
+
+        MemberSession s = sessions.acquire(null, null, "/caller",
+                null, new WorktreeRequest("fleetd-425c", null));
+
+        assertEquals("b", s.profile(),
+                "a is quarantined, so the unqualified worktree spawn must route to b");
+        FakeWorktrees.OverlayCall overlay = worktrees.lastOverlay();
+        assertNotNull(overlay, "overlayParity must have been called");
+        assertEquals(List.of("b.mcp.json"), overlay.requested(),
+                "parityOverlay must be provisioned for b — the profile actually spawned, never a's, "
+                        + "the quarantined pool-first entry");
+        FakeWorktrees.RepoRootCall repoRootCall = worktrees.repoRootCalls().getLast();
+        assertTrue(repoRootCall.cwd().contains("/repo/b"),
+                "repoRoot must be resolved through b's effectiveCwd, not a's: " + repoRootCall.cwd());
+    }
+
+    /**
+     * fleetd #425 rework, round 2: this is the exact probe that found round 1's maxLoad
+     * regression. One dev profile ("a") is configured with {@code maxLoad: 1} and a liveCount
+     * pinned at 1 — permanently at cap — under {@code PlacementPolicies.fixed()}, the default
+     * policy, which deliberately never evaluates {@code maxLoad} during automatic selection (see
+     * {@code CompositePeerLauncher}'s own javadoc on {@code place}/{@code FixedPlacementPolicy}).
+     *
+     * <p>Round 1 resolved {@code acquireWithWorktree}'s profile through
+     * {@code launcher.routedProfileFor(memberRole)} and then fed that name back into
+     * {@code launcher.spawn(SpawnRequest)} as an EXPLICIT profile. Naming a profile explicitly
+     * takes {@code CompositePeerLauncher.spawn}'s THROWING branch, which calls
+     * {@code enforceMaxLoad} — so the worktree path died with a {@code PlacementException} while
+     * the exact same unqualified request, with no worktree, still spawned cleanly through the
+     * routing branch that never checks {@code maxLoad} at all. One intent, two different answers,
+     * depending only on whether a worktree was asked for — the #425 shape, moved to a different
+     * filter instead of closed.
+     *
+     * <p>This test does not hardcode which of the two outcomes is correct — whether an unqualified
+     * spawn SHOULD respect {@code maxLoad} is fleetd #435, a separate ticket. It only asserts that
+     * the WITH-worktree and WITHOUT-worktree paths agree: both spawn on the same profile, or both
+     * fail with the same exception type and message. That way this test stays correct however
+     * #435 is eventually resolved, and only breaks if the two paths disagree again.
+     */
+    @Test
+    void unqualifiedAcquireAgreesWithAndWithoutAWorktreeWhenTheOnlyProfileIsAtMaxLoad() {
+        Map<String, FleetConfig.Profile> profiles = new LinkedHashMap<>();
+        profiles.put("a", new FleetConfig.Profile("a", "http://gx00.gw:8000", "coder-a", null,
+                "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, "/repo/a", List.of("a.mcp.json"),
+                null, null, null, null, 1.0f, 1));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "a", _ -> null);
+        // liveCount pinned at 1 for "a", exactly matching maxLoad — "a" is permanently at cap,
+        // regardless of how many times either branch below actually spawns.
+        PeerLauncher launcher = new CompositePeerLauncher(List.of(adapter), "a", profiles,
+                PlacementPolicies.fixed(), name -> "a".equals(name) ? 1 : 0);
+
+        Object without = attemptAcquire(() ->
+                new SessionManager(launcher, new FakeWorktrees(), () -> 0L)
+                        .acquire(null, null, "/caller", null));
+        Object with = attemptAcquire(() ->
+                new SessionManager(launcher, new FakeWorktrees(), () -> 0L)
+                        .acquire(null, null, "/caller", null, new WorktreeRequest("fleetd-425-maxload", null)));
+
+        assertEquals(without, with, "an unqualified spawn on a profile at maxLoad must agree "
+                + "whether or not a worktree was requested — no condition may become newly fatal "
+                + "on the worktree path alone (fleetd #425 rework, round 2)");
+    }
+
+    /**
+     * fleetd #425 rework, round 4: the exact regression a mutation test found that 186 green tests
+     * missed — {@code acquireWithWorktree} dropping the {@link
+     * dev.ltms.fleet.placement.PlacementDecision} it already resolved via {@code launcher.place},
+     * and letting the unqualified spawn re-run placement a second time (a blank-profile {@code
+     * launcher.spawn(spawnReq)}) instead of carrying that decision forward via {@code
+     * launcher.spawn(spawnReq, decision)}. Every earlier test in this file uses {@code
+     * PlacementPolicies.fixed()}, which returns the same answer on every {@code select()} call, so
+     * dropping the decision is invisible under it — two {@code select()} calls simply agree by
+     * accident. {@code PlacementPolicies.roundRobin()} is deterministic AND stateful: its {@code
+     * select()} advances an internal index on every call, so two consecutive calls for the SAME
+     * spawn (one from {@code place()} to provision the worktree, a second from a dropped-decision
+     * blank-profile {@code spawn(spawnReq)}) land on DIFFERENT profiles from a two-profile pool —
+     * index 0 ("a"), then index 1 ("b").
+     *
+     * <p>This test does not hardcode which profile wins — asserting one specific name would pass
+     * for the wrong reason the moment the rotation order changes (round-4 brief invariant 3). It
+     * asserts AGREEMENT instead: whichever profile the worktree's parity overlay was provisioned
+     * for must be the SAME profile the member actually spawned on. Each profile's overlay list is
+     * named after the profile itself ({@code "a.mcp.json"}/{@code "b.mcp.json"}), so comparing the
+     * recorded overlay against {@code s.profile() + ".mcp.json"} checks agreement without ever
+     * naming an expected winner.
+     */
+    @Test
+    void acquireWithWorktreeSpawnsOnTheSameProfileItProvisionedTheWorktreeForUnderARotatingPolicy() {
+        Map<String, FleetConfig.Profile> profiles = new LinkedHashMap<>();
+        profiles.put("a", new FleetConfig.Profile("a", "http://gx00.gw:8000", "coder-a", null,
+                "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, "/repo/a", List.of("a.mcp.json")));
+        profiles.put("b", new FleetConfig.Profile("b", "http://gx00.gw:8000", "coder-b", null,
+                "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, "/repo/b", List.of("b.mcp.json")));
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), profiles, "a", _ -> null);
+        // roundRobin is deterministic AND stateful: the first select() call picks index 0 ("a"),
+        // and the SAME policy instance's second select() call (reached only if the
+        // PlacementDecision is dropped) picks index 1 ("b") — the two-call disagreement this test
+        // needs to make a dropped decision observable, rather than merely probable.
+        PeerLauncher launcher = new CompositePeerLauncher(List.of(adapter), "a", profiles,
+                PlacementPolicies.roundRobin(), _ -> 0);
+
+        FakeWorktrees worktrees = new FakeWorktrees();
+        SessionManager sessions = new SessionManager(launcher, worktrees, () -> 0L);
+
+        MemberSession s = sessions.acquire(null, null, "/caller",
+                null, new WorktreeRequest("fleetd-425-round4", null));
+
+        FakeWorktrees.OverlayCall overlay = worktrees.lastOverlay();
+        assertNotNull(overlay, "overlayParity must have been called");
+        assertEquals(List.of(s.profile() + ".mcp.json"), overlay.requested(),
+                "the worktree must be provisioned for the SAME profile the member actually spawned "
+                        + "on — under a rotating policy, dropping the PlacementDecision makes the "
+                        + "second, spawn-time select() call disagree with the first, place()-time "
+                        + "call, so the member ends up on a profile whose worktree (repoRoot/parity "
+                        + "overlay) was built for a DIFFERENT profile (fleetd #425 rework, round 4)");
+    }
+
+    /**
+     * Reduce one {@code acquire(...)} attempt to a value comparable across the with-worktree and
+     * without-worktree paths: the spawned profile name on success, or the thrown exception's class
+     * and message on failure. Comparing THIS — instead of asserting "both spawn" or "both throw" as
+     * a hardcoded direction — is what keeps {@link
+     * #unqualifiedAcquireAgreesWithAndWithoutAWorktreeWhenTheOnlyProfileIsAtMaxLoad} valid whichever
+     * way fleetd #435 eventually resolves whether an unqualified spawn should respect maxLoad.
+     */
+    private static Object attemptAcquire(Supplier<MemberSession> call) {
+        try {
+            return "spawned:" + call.get().profile();
+        } catch (RuntimeException e) {
+            return "threw:" + e.getClass().getName() + ":" + e.getMessage();
+        }
+    }
 }