Merge #436: make fixed placement honour maxLoad (fleetd #435)

fixed was the one automatic policy that ignored maxLoad, and it is the
default for an absent placement: key. It now gates on the same shared
atCap predicate weighted and round-robin use, and falls through to the
next candidate rather than refusing.

Verified here: 1555 tests green (main was 1548, +7 new methods), 0
compile errors. Mutation battery, control 113 green:
  - atCap boundary >= -> >        KILLED (15 tests, across all 3 policies)
  - drop cap check in fixed walk  KILLED (3 tests)
  - weightExcluded -> false       KILLED (2 pre-existing tests)

Neither live host is affected today: both run placement: weighted (Mac
fleetd.yaml:210, fleet01 fleetd.yaml:172), and weighted already skipped
at-cap candidates via PlacementPolicyUtil.available. This aligns fixed
with the other policies and with maxLoad's documented contract.

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

@@ -5,14 +5,12 @@ import java.util.List;
 
 /**
  * Backward-compatible placement: an unqualified spawn always resolves to the configured default
- * profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps
- * ({@code maxLoad}) so that a pre-existing config behaves identically after upgrade — capacity
- * gating for automatic placement is deliberately out of scope for {@code fixed}, exactly as it
- * always has been. Reachability is a narrower exception (fleetd #315, below): a profile is never
- * checked for reachability up front, only skipped once it has already failed in <em>this same</em>
- * spawn call's retry loop — see the unreachable case below.
+ * profile, exactly as {@code CompositePeerLauncher} did before CB-518. Reachability is a narrower
+ * exception (fleetd #315, below): a profile is never checked for reachability up front, only
+ * skipped once it has already failed in <em>this same</em> spawn call's retry loop — see the
+ * unreachable case below.
  *
- * <p>Five exceptions walk past the default instead of returning it unconditionally:
+ * <p>Six exceptions walk past the default instead of returning it unconditionally:
  * <ul>
  *   <li>Quarantine (CB-578 stage B): a quarantined default is a credential that just refused on
  *       a usage limit, not a transient capacity or reachability concern.
@@ -20,14 +18,24 @@ import java.util.List;
  *       ({@code BackendOutagePolicy}) — a separate, shorter-lived source from quarantine. When a
  *       profile is both quarantined and cooling off, only the quarantine reason is reported
  *       (exhaustion takes priority), matching {@code CompositePeerLauncher}'s explicit-spawn order.
+ *   <li>At cap (fleetd #435): a profile whose live count has reached its {@code maxLoad}
+ *       ({@link PlacementPolicyUtil#atCap}) — a documented, unconditional capacity limit (see
+ *       {@code FleetConfig.Profile#maxLoad}), so {@code fixed} must gate on it exactly as {@code
+ *       weighted}/{@code round-robin} already do via {@link PlacementPolicyUtil#available}. Before
+ *       this fix {@code fixed} built its own {@link PlacementCandidate} for the default with {@code
+ *       maxLoad} forced to {@code null}, so a capped default was chosen anyway on every unqualified
+ *       spawn — the cap was advisory, not enforced, for the one placement policy every config uses
+ *       by default. Reported only when quarantine and cooling off are both absent, matching {@code
+ *       CompositePeerLauncher}'s explicit-spawn check order (quarantine, then cooling off, then max
+ *       load, then model-off).
  *   <li>Model off (fleetd #422): a profile whose {@code model:} the operator has turned off in
  *       {@code models.allow:} — an operator decision, never a backend-reported outage, so it is a
- *       FOURTH, independent source from both quarantine and cooling off (never merged with either),
- *       exactly as {@code CompositePeerLauncher.enforceModelEnabled} and {@link
- *       PlacementPolicyUtil#available} treat it. When a profile is model-off <em>and</em> quarantined
- *       or cooling off, only the quarantine/cooling-off reason is reported — those still take
- *       priority, matching {@code CompositePeerLauncher}'s explicit-spawn check order (quarantine,
- *       then cooling off, then max load, then model-off).
+ *       fifth, independent source from quarantine, cooling off, and at-cap (never merged with any
+ *       of them), exactly as {@code CompositePeerLauncher.enforceModelEnabled} and {@link
+ *       PlacementPolicyUtil#available} treat it. When a profile is model-off <em>and</em> quarantined,
+ *       cooling off, or at cap, only the higher-priority reason is reported, matching {@code
+ *       CompositePeerLauncher}'s explicit-spawn check order (quarantine, then cooling off, then max
+ *       load, then model-off).
  *   <li>Unreachable (fleetd #315): {@code CompositePeerLauncher.spawn} retries a failed candidate
  *       on the next one and rebuilds the {@link PlacementContext} so {@code ctx.unreachable()}
  *       names every profile that already failed with {@code PeerUnreachableException} in this same
@@ -40,10 +48,10 @@ import java.util.List;
  *       {@code weighted}/{@code round-robin} skip it — an explicit {@code fleet_spawn} naming
  *       the profile is unaffected, only this automatic fallback walk.
  * </ul>
- * A fleet where nothing is ever quarantined, cooling off, model-off, unreachable, or weight-0 never
- * exercises any of these paths, so today's behaviour is unchanged — in particular, the very first
- * selection of a spawn call always sees an empty {@code unreachable} set, so the first choice is
- * untouched.
+ * A fleet where nothing is ever quarantined, cooling off, at cap, model-off, unreachable, or
+ * weight-0 never exercises any of these paths, so today's behaviour is unchanged — in particular,
+ * the very first selection of a spawn call always sees an empty {@code unreachable} set, so the
+ * first choice is untouched.
  */
 final class FixedPlacementPolicy implements PlacementPolicy {
 
@@ -51,13 +59,14 @@ final class FixedPlacementPolicy implements PlacementPolicy {
     public PlacementCandidate select(PlacementContext ctx) {
         String d = ctx.defaultProfile();
         if (d != null && !d.isBlank() && !ctx.quarantined().contains(d) && !ctx.coolingOff().contains(d)
-                && !ctx.modelOff().contains(d) && !ctx.unreachable().contains(d) && !weightExcluded(ctx, d)) {
+                && !ctx.modelOff().contains(d) && !ctx.unreachable().contains(d) && !weightExcluded(ctx, d)
+                && !capExcluded(ctx, d)) {
             return new PlacementCandidate(d, null, 1.0f, null);
         }
         for (PlacementCandidate c : ctx.candidates()) {
             if (!ctx.quarantined().contains(c.profile()) && !ctx.coolingOff().contains(c.profile())
                     && !ctx.modelOff().contains(c.profile()) && !ctx.unreachable().contains(c.profile())
-                    && !c.excluded()) {
+                    && !c.excluded() && !PlacementPolicyUtil.atCap(ctx, c)) {
                 return new PlacementCandidate(c.profile(), null, c.weight(), c.maxLoad());
             }
         }
@@ -66,14 +75,18 @@ final class FixedPlacementPolicy implements PlacementPolicy {
             // Exhaustion quarantine takes priority: reported only when quarantine is absent, so the
             // message never claims "cooling off" for a profile that is really backend-exhausted.
             boolean dCoolingOff = !dQuarantined && ctx.coolingOff().contains(d);
-            // fleetd #422: model-off is a fourth, independent source (an operator decision) — but
-            // quarantine/cooling-off still take priority when more than one applies, matching
+            // fleetd #435: at-cap sits between cooling off and model-off, matching
+            // CompositePeerLauncher's explicit-spawn check order (quarantine, cooling off, max load,
+            // then model-off) — reported only when quarantine/cooling-off are both absent.
+            boolean dAtCap = !dQuarantined && !dCoolingOff && capExcluded(ctx, d);
+            // fleetd #422: model-off is a fifth, independent source (an operator decision) — but
+            // quarantine/cooling-off/at-cap still take priority when more than one applies, matching
             // CompositePeerLauncher's explicit-spawn check order (quarantine, cooling off, max load,
             // then model-off).
-            boolean dModelOff = !dQuarantined && !dCoolingOff && ctx.modelOff().contains(d);
+            boolean dModelOff = !dQuarantined && !dCoolingOff && !dAtCap && ctx.modelOff().contains(d);
             boolean dUnreachable = ctx.unreachable().contains(d);
             boolean dWeightExcluded = weightExcluded(ctx, d);
-            if (dQuarantined || dCoolingOff || dModelOff || dUnreachable || dWeightExcluded) {
+            if (dQuarantined || dCoolingOff || dAtCap || dModelOff || dUnreachable || dWeightExcluded) {
                 List<String> reasons = new ArrayList<>();
                 if (dQuarantined) {
                     reasons.add("is quarantined (backend exhausted)");
@@ -81,6 +94,11 @@ final class FixedPlacementPolicy implements PlacementPolicy {
                 if (dCoolingOff) {
                     reasons.add("is cooling off after repeated backend errors");
                 }
+                if (dAtCap) {
+                    PlacementCandidate c = candidateFor(ctx, d);
+                    int live = ctx.liveCount().apply(d);
+                    reasons.add("is at maxLoad (" + live + " live >= " + c.maxLoad() + " cap)");
+                }
                 if (dModelOff) {
                     reasons.add("names a model the operator has turned off in models.allow");
                 }
@@ -96,18 +114,38 @@ final class FixedPlacementPolicy implements PlacementPolicy {
         }
         if (!ctx.candidates().isEmpty()) {
             throw new PlacementException("all worker profiles are excluded from automatic "
-                    + "selection (quarantined, cooling off, model-off, unreachable, or weight-0)");
+                    + "selection (quarantined, cooling off, at cap, model-off, unreachable, or weight-0)");
         }
         throw new PlacementException("no worker profiles configured");
     }
 
     /** Whether {@code profile} carries {@code weight <= 0} (CB-554) among {@code ctx}'s candidates. */
     private static boolean weightExcluded(PlacementContext ctx, String profile) {
+        PlacementCandidate c = candidateFor(ctx, profile);
+        return c != null && c.excluded();
+    }
+
+    /**
+     * Whether {@code profile} has reached its {@code maxLoad} cap (fleetd #435), using the shared
+     * {@link PlacementPolicyUtil#atCap} definition — the same one {@code weighted}/{@code
+     * round-robin} already consult via {@link PlacementPolicyUtil#available}. Looked up by name,
+     * the same way {@link #weightExcluded} is: the default fast path above builds its own {@link
+     * PlacementCandidate} with {@code maxLoad} forced to {@code null} (it carries no cap of its
+     * own), so the candidate actually configured for {@code profile} has to be found in {@code
+     * ctx.candidates()} first.
+     */
+    private static boolean capExcluded(PlacementContext ctx, String profile) {
+        PlacementCandidate c = candidateFor(ctx, profile);
+        return c != null && PlacementPolicyUtil.atCap(ctx, c);
+    }
+
+    /** The configured candidate named {@code profile} in {@code ctx}, or {@code null} if none. */
+    private static PlacementCandidate candidateFor(PlacementContext ctx, String profile) {
         for (PlacementCandidate c : ctx.candidates()) {
             if (c.profile().equals(profile)) {
-                return c.excluded();
+                return c;
             }
         }
-        return false;
+        return null;
     }
 }

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

@@ -11,14 +11,29 @@ final class PlacementPolicyUtil {
     private PlacementPolicyUtil() {
     }
 
+    /**
+     * True when {@code c} has reached its {@code maxLoad} cap: {@code liveCount(c.profile()) >=
+     * c.maxLoad()}. A {@code null} maxLoad means unlimited, so it is never at cap.
+     *
+     * <p>Extracted as the single shared definition of "at cap" (fleetd #435): before this fix it
+     * was computed inline in both {@link #available} and {@link #emptyException}, and {@code
+     * FixedPlacementPolicy} — not a caller of either — quietly kept its own {@code select} free of
+     * any cap check at all, so a capped default profile was chosen anyway under the default
+     * placement policy. Every automatic policy must call this, not re-derive it.
+     */
+    static boolean atCap(PlacementContext ctx, PlacementCandidate c) {
+        Integer cap = c.maxLoad();
+        return cap != null && ctx.liveCount().apply(c.profile()) >= cap;
+    }
+
     /**
      * Candidates that are not weight-excluded (CB-554: explicit {@code weight <= 0}, checked
      * first because it is a static config choice rather than transient state), not
      * known-unreachable, not quarantined (CB-578 stage B), not cooling off after repeated backend
      * errors (fleetd #201 Unit 5 — a separate, shorter-lived source from quarantine), not naming a
      * model the operator has turned off (fleetd #422 — a third, independent source: an operator
-     * decision, never a backend-reported outage), and have not reached their maxLoad. A {@code
-     * null} maxLoad means unlimited.
+     * decision, never a backend-reported outage), and have not reached their maxLoad (see {@link
+     * #atCap}). A {@code null} maxLoad means unlimited.
      */
     static List<PlacementCandidate> available(PlacementContext ctx) {
         List<PlacementCandidate> out = new ArrayList<>();
@@ -26,16 +41,10 @@ final class PlacementPolicyUtil {
             if (c.excluded() || ctx.unreachable().contains(c.profile())
                     || ctx.quarantined().contains(c.profile())
                     || ctx.coolingOff().contains(c.profile())
-                    || ctx.modelOff().contains(c.profile())) {
+                    || ctx.modelOff().contains(c.profile())
+                    || atCap(ctx, c)) {
                 continue;
             }
-            Integer cap = c.maxLoad();
-            if (cap != null) {
-                int live = ctx.liveCount().apply(c.profile());
-                if (live >= cap) {
-                    continue;
-                }
-            }
             out.add(c);
         }
         return out;
@@ -59,7 +68,6 @@ final class PlacementPolicyUtil {
         int coolingOff = 0;
         int modelOff = 0;
         for (PlacementCandidate c : ctx.candidates()) {
-            Integer cap = c.maxLoad();
             if (c.excluded()) {
                 weightExcluded++;
             } else if (ctx.quarantined().contains(c.profile())) {
@@ -70,7 +78,7 @@ final class PlacementPolicyUtil {
                 modelOff++;
             } else if (ctx.unreachable().contains(c.profile())) {
                 unreachable++;
-            } else if (cap != null && ctx.liveCount().apply(c.profile()) >= cap) {
+            } else if (atCap(ctx, c)) {
                 atCap++;
             }
         }

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

@@ -536,6 +536,31 @@ class CompositePeerLauncherTest {
         assertEquals("claude", h.profile(), "the returned handle carries the resolved default profile");
     }
 
+    /**
+     * fleetd #435: {@code FixedPlacementPolicy} — the default placement policy every config uses
+     * unless {@code placement:} is set — never consulted {@code maxLoad}, so an unqualified spawn
+     * (a blank profile, the normal delegation path) landed on a capped default anyway. Measured on
+     * 7667727: a single dev profile at {@code maxLoad: 1} with 1 live, under {@code fixed()},
+     * returned "SPAWNED on profile=a". This test goes through {@code CompositePeerLauncher.spawn}
+     * with a blank profile, not the policy in isolation, so it proves the caller actually reaches
+     * the fixed default's new cap check rather than only the {@code select} method.
+     */
+    @Test
+    void fixedPolicyGatesDefaultProfileAtMaxLoadOnUnqualifiedSpawn() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, FleetConfig.Profile> profiles = ordered(
+                "a", stubWorker("a", 1.0f, 1),
+                "b", stubWorker("b", 1.0f, null));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "a", Set.of());
+        CompositePeerLauncher composite = new CompositePeerLauncher(
+                List.of(adapter), "a", profiles, PlacementPolicies.fixed(), name -> "a".equals(name) ? 1 : 0);
+
+        PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
+        assertEquals("b", h.profile(),
+                "the default profile a is at maxLoad, so fixed placement must fall through to b");
+        assertEquals(0, adapter.spawnCount("a"), "a is never spawned — it is already at cap");
+    }
+
     @Test
     void weightedPolicyGatesProfileAtMaxLoad() {
         FakeHerdr herdr = new FakeHerdr();

fleetd/src/test/java/dev/ltms/fleet/placement/PlacementPolicyTest.java

@@ -461,6 +461,93 @@ class PlacementPolicyTest {
         assertTrue(e.getMessage().contains("weight 0"), e.getMessage());
     }
 
+    // --- fleetd #435: FixedPlacementPolicy must consult maxLoad too, at BOTH filter sites -------
+
+    /**
+     * The default-profile fast path must skip a capped default. Measured on 7667727 before this
+     * fix: a single dev profile at {@code maxLoad: 1} with 1 live, under {@code fixed()}, still
+     * returned "SPAWNED on profile=a" — the cap was advisory for every unqualified spawn.
+     */
+    @Test
+    void fixedSkipsCappedDefault() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a", 1.0f, null),
+                        PlacementCandidate.profile("b", 1.0f, 1)),
+                name -> "b".equals(name) ? 1 : 0, Set.of(), Set.of(), Set.of());
+        assertEquals("a", policy.select(ctx).profile(),
+                "the default 'b' is at its maxLoad cap, so fixed falls through to the free candidate 'a'");
+    }
+
+    /**
+     * The fallback walk must skip a capped candidate too — exercised independently of the
+     * default-profile fast path by using no default at all, so this is the only filter that runs.
+     */
+    @Test
+    void fixedFallbackWalkSkipsCappedCandidate() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext(null,
+                List.of(PlacementCandidate.profile("a", 1.0f, 1),
+                        PlacementCandidate.profile("b", 1.0f, null)),
+                name -> "a".equals(name) ? 1 : 0, Set.of(), Set.of(), Set.of());
+        assertEquals("b", policy.select(ctx).profile(),
+                "candidate 'a' is at its maxLoad cap, so the fallback walk skips it and picks 'b'");
+    }
+
+    @Test
+    void fixedThrowsWhenDefaultAndEveryCandidateAtCap() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a", 1.0f, 1),
+                        PlacementCandidate.profile("b", 1.0f, 1)),
+                _ -> 1, Set.of(), Set.of(), Set.of());
+        PlacementException e = assertThrows(PlacementException.class, () -> policy.select(ctx));
+        assertTrue(e.getMessage().contains("maxLoad"), "message names the cap: " + e.getMessage());
+        assertTrue(e.getMessage().contains("1 cap"), "message names the cap value: " + e.getMessage());
+        assertFalse(e.getMessage().contains("quarantined"), "must not read like quarantine: " + e.getMessage());
+        assertFalse(e.getMessage().contains("turned off"), "must not read like model-off: " + e.getMessage());
+    }
+
+    /** The mirror: an uncapped default is still chosen, so the new term cannot exclude everything. */
+    @Test
+    void fixedStillReturnsUncappedDefault() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a", 1.0f, null),
+                        PlacementCandidate.profile("b", 1.0f, null)),
+                noSessions(), Set.of(), Set.of(), Set.of());
+        assertEquals("b", policy.select(ctx).profile(), "an uncapped default is returned unconditionally");
+    }
+
+    /** CB-585: an explicit {@code maxLoad: 0} on the default caps it at zero live members. */
+    @Test
+    void fixedSkipsMaxLoadZeroDefaultEvenWithZeroLiveWorkers() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a", 1.0f, null),
+                        PlacementCandidate.profile("b", 1.0f, 0)),
+                _ -> 0, Set.of(), Set.of(), Set.of());
+        assertEquals("a", policy.select(ctx).profile(),
+                "the default 'b' has maxLoad 0, so it is already at its cap with nobody live");
+    }
+
+    /**
+     * Quarantine still wins when a profile is both quarantined and at cap (mirrors {@code
+     * fixedReportsQuarantineNotModelOffWhenBothApply}'s priority over model-off).
+     */
+    @Test
+    void fixedReportsQuarantineNotAtCapWhenBothApply() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a", 1.0f, 1),
+                        PlacementCandidate.profile("b", 1.0f, 1)),
+                _ -> 1, Set.of(), Set.of("a", "b"), Set.of());
+        PlacementException e = assertThrows(PlacementException.class, () -> policy.select(ctx));
+        assertTrue(e.getMessage().contains("quarantined"), e.getMessage());
+        assertFalse(e.getMessage().contains("maxLoad"),
+                "quarantine takes priority over at-cap in the message: " + e.getMessage());
+    }
+
     @Test
     void unknownPolicyNameThrows() {
         assertThrows(IllegalArgumentException.class, () -> PlacementPolicies.fromName("random"));