CB-598: track reminder counts per pending item, not per lead per source

Work that arrived during the ~15s push_backoff_ms window between two
ticks landed in the pending map before the next tick's start-of-tick
snapshot, so a shared per-lead-per-source counter (carried forward via
scheduleNext(lead, count+1, ...)) already treated it as exhausted
backlog even though no nudge had ever named it. ReplyPushLoop.tick now
recomputes each source's reminder count fresh every tick as the
minimum nudge count among that source's currently pending items, so a
freshly-arrived item (count 0) keeps its source eligible regardless of
how depleted an older, still-undrained sibling's count is. decide()
itself is unchanged.

bridged/bridged.example.yaml

@@ -88,28 +88,15 @@ bind:
 #   backoffMs: 60000
 #   quietNudgeCap: 3
 
-# Fleet health detection is dormant unless enabled (CB-573). It reads one whole-fleet agent list
-# per tick.
-#   intervalSeconds → how often a tick runs (default 30). ENFORCED floor of 15: the code computes
-#                      Math.max(15, intervalSeconds), so a lower value is silently raised, not
-#                      rejected.
-#   workingSuspectAfterSeconds, paneProbeIntervalSeconds → accepted and parsed, but NOT YET READ by
-#                      anything — the dormant monitor only consumes intervalSeconds today (CB-573
-#                      shipped ahead of the evidence publishers these two knobs are for). Setting
-#                      them changes nothing right now, and no minimum is enforced on either, because
-#                      nothing reads them to enforce one. They exist so a later build can start
-#                      honouring them without another config-shape change.
-#   notifications.mode → "webhook" flips what bridge_list REPORTS (healthCoverage: "full" instead
-#                      of "detection-only") — it does NOT make bridged send any webhook call; no
-#                      delivery mechanism is implemented yet. Any other value, or omitting the
-#                      block, reports "detection-only".
+# Fleet health detection is dormant unless enabled. It reads one whole-fleet agent list per tick.
+# It can run without a webhook; bridge_list then reports healthCoverage: detection-only.
 # health:
 #   enabled: true
-#   intervalSeconds: 30
-#   workingSuspectAfterSeconds: 600
-#   paneProbeIntervalSeconds: 60
+#   intervalSeconds: 30              # minimum 15
+#   workingSuspectAfterSeconds: 600  # minimum 300
+#   paneProbeIntervalSeconds: 60     # minimum 60
 #   notifications:
-#     mode: disabled
+#     mode: disabled                 # disabled (default) or webhook
 
 # herdr Unix socket. Omit to use the client default
 # (${HERDR_SOCKET_PATH:-~/.config/herdr/herdr.sock}).
@@ -299,11 +286,6 @@ placement: weighted
 #               / 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.
-#               EXCEPT `fleet.leaders`: Bridged.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"
-#               with nothing in the deferred list — but has NO effect until you restart. Treat it
-#               as deferred in practice, even though today's reload output does not say so.
 #   DEFERRED  → accepted into the new config, but the wiring built at startup keeps the old value
 #               until you restart: `lifecycle:`, `leadHeartbeat:`, `guard:`, `worktreeRoot:`,
 #               `spawnReadyTimeoutMs` / `spawnReadyPollMs`, `quarantineCooldownSeconds` (CB-578
@@ -386,12 +368,6 @@ fleet:
   # An auto-launched lead is NOT a member: it gets no worker reply charter, is never registered with
   # the session lifecycle (the idle reaper would kill your orchestrator), and stays on the
   # subscription — ANTHROPIC_BASE_URL/AUTH_TOKEN are stripped from its env whatever the profile says.
-  #
-  # GET THE `tab:` VALUE RIGHT. A pane that does not match any configured `tab:` (a typo, a renamed
-  # tab, a pane no entry names at all) is not recognised as a lead — it resolves as an ordinary
-  # WORKER instead, silently, and every orchestration call it makes (spawn/stop/send/drain) is
-  # refused. There is no error at startup for this: an unmatched pane is simply not a lead. If your
-  # primary suddenly can't spawn or send, check this section first.
   # leaders:
   #   opus-5.0:
   #     profile: opus                # omit to never create this lead, only recognise it
@@ -447,15 +423,10 @@ guard:
 #   idleTtlSeconds     → reap READY/DONE sessions idle longer than this (never BUSY/SPAWNING)
 #   contextCap         → force-release a session after this many delegated turns
 #   drainTimeoutSeconds → seconds to wait for BUSY sessions on shutdown before forced teardown
-#   clearAfterTurn     → whether a reusable worker discards its conversation context after every
-#                        completed delegated turn (default false). Works for claude-code workers
-#                        only — any other peer kind (e.g. opencode) logs "context reset is
-#                        unsupported for peer kind …" once and the reset is a no-op.
 # lifecycle:
 #   idleTtlSeconds: 300
 #   contextCap: 10
 #   drainTimeoutSeconds: 5
-#   clearAfterTurn: false
 
 # Durable reply delivery (CB-307 Stage 2). OMIT this block entirely to keep the default
 # in-memory, soft-state reply inbox (late worker replies are held only until a daemon bounce).

bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java

@@ -967,12 +967,8 @@ public record BridgedConfig(
     /**
      * Top-level keys this version understands. Used only to warn about the rest — see
      * {@link #warnUnknownTopLevelKeys}. Keep in step with the record components.
-     *
-     * <p>Package-private (not {@code private}) so a test can assert every key here is documented in
-     * {@code bridged.example.yaml} — the only committed description of the config schema, since
-     * {@code bridged.yaml} itself is gitignored.
      */
-    static final Set<String> KNOWN_TOP_LEVEL_KEYS = Set.of(
+    private static final Set<String> KNOWN_TOP_LEVEL_KEYS = Set.of(
             "bind", "herdrSocket", "profiles", "guard", "worktreeRoot",
             "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs", "broker", "primary", "fleet",
             "leadHeartbeat", "health", "placement", "auth", "configReload", "quarantineCooldownSeconds");

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

@@ -15,7 +15,6 @@ import dev.ltms.bridged.msg.MessageService;
 import dev.ltms.bridged.msg.Rendezvous;
 import dev.ltms.bridged.peer.PeerUnreachableException;
 import dev.ltms.bridged.placement.BackendQuarantine;
-import dev.ltms.bridged.placement.PlacementException;
 import dev.ltms.bridged.session.SessionManager;
 import dev.ltms.bridged.session.MemberSession;
 import dev.ltms.bridged.session.WorktreeRequest;
@@ -693,10 +692,6 @@ public final class BridgeMcp {
             return text(json(memberView(member)));
         } catch (GuardException e) {
             return error("subscription boundary: " + e.getMessage());
-        } catch (PlacementException e) {
-            // CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — distinct
-            // from "profile does not exist" below.
-            return error("no capacity: " + e.getMessage());
         } catch (IllegalArgumentException e) {
             return error(e.getMessage()); // unknown / no-default profile, or a refused resumeSessionId
         } catch (PeerUnreachableException e) {

bridged/src/main/java/dev/ltms/bridged/msg/ReplyPushLoop.java

@@ -66,8 +66,13 @@ public final class ReplyPushLoop {
     private final long backoffMs;
     private final Metrics metrics; // CB-512: nullable — no registry in unit tests
 
-    /** Worker targets with a reply queued, and the lead to nudge about it, keyed by target. */
-    private final ConcurrentHashMap<String, String> pendingReplies = new ConcurrentHashMap<>();
+    /**
+     * Worker targets with a reply queued, keyed by target. Each entry carries its own nudge
+     * count (CB-598) rather than sharing one counter per lead per source: a target's count only
+     * ever reflects nudges that actually named that target, so a target that joins while the
+     * schedule is already deep into another target's reminders still reads as fresh.
+     */
+    private final ConcurrentHashMap<String, ReplyEntry> pendingReplies = new ConcurrentHashMap<>();
     /** Tickets that have gone terminal but not yet been polled, keyed by ticket. */
     private final ConcurrentHashMap<String, PendingTicket> pendingTickets = new ConcurrentHashMap<>();
     /** CB-590: leads with an active combined reminder schedule (replies and/or tickets). */
@@ -116,10 +121,10 @@ public final class ReplyPushLoop {
         Set<String> result = new HashSet<>();
         for (var entry : pendingReplies.entrySet()) {
             String target = entry.getKey();
-            String owningLead = entry.getValue();
-            if (!lead.equals(owningLead)) continue;
+            ReplyEntry owning = entry.getValue();
+            if (!lead.equals(owning.lead())) continue;
             if (inbox.peek(target).isEmpty()) {
-                pendingReplies.remove(target, owningLead);
+                pendingReplies.remove(target, owning);
                 continue;
             }
             result.add(target);
@@ -127,8 +132,15 @@ public final class ReplyPushLoop {
         return result;
     }
 
-    /** A ticket awaiting collection: which lead to nudge, and whether it ended in failure. */
-    private record PendingTicket(String ticket, String lead, boolean failed) {
+    /** A pending reply target: which lead to nudge, and how many nudges have named it so far. */
+    private record ReplyEntry(String lead, int nudgeCount) {
+    }
+
+    /**
+     * A ticket awaiting collection: which lead to nudge, whether it ended in failure, and how
+     * many nudges have named it so far (CB-598 — tracked per ticket, not per lead per source).
+     */
+    private record PendingTicket(String ticket, String lead, boolean failed, int nudgeCount) {
     }
 
     /** Tickets still pending for {@code lead}, snapshotted fresh for one tick. */
@@ -142,6 +154,41 @@ public final class ReplyPushLoop {
                 .collect(Collectors.toUnmodifiableSet());
     }
 
+    /**
+     * The reply-source reminder count {@link #decide} should see for {@code lead} on this tick:
+     * the <em>minimum</em> nudge count among the reply targets currently pending for it (CB-598).
+     *
+     * <p>Before this, the count passed to {@code decide} was a single counter carried forward
+     * across scheduled ticks ({@code scheduleNext(lead, count + 1, ...)}), incremented whenever
+     * the source had <em>any</em> pending work — not tied to which target that work was. A target
+     * that joined while an older target's count was already near the cap inherited that count on
+     * its very next tick, even though no nudge had ever named it. Taking the minimum over what is
+     * actually pending now means a fresh target (count 0) keeps the source eligible regardless of
+     * how many times an older, still-undrained target has already been nudged; that older target
+     * keeps riding along in the combined nudge text without spending any more of its own budget
+     * (see {@link #bumpNudgeCounts}). Returns 0 when nothing is pending — {@link #decide} never
+     * consults the count in that case, since {@code hasReplyWork} is false.
+     */
+    private int minReplyNudgeCountFor(String lead) {
+        int min = Integer.MAX_VALUE;
+        for (String target : pendingReplyTargetsFor(lead)) {
+            ReplyEntry entry = pendingReplies.get(target);
+            if (entry != null) {
+                min = Math.min(min, entry.nudgeCount());
+            }
+        }
+        return min == Integer.MAX_VALUE ? 0 : min;
+    }
+
+    /** As {@link #minReplyNudgeCountFor}, for the ticket source. */
+    private int minTicketNudgeCountFor(String lead) {
+        int min = Integer.MAX_VALUE;
+        for (PendingTicket ticket : pendingTicketsFor(lead)) {
+            min = Math.min(min, ticket.nudgeCount());
+        }
+        return min == Integer.MAX_VALUE ? 0 : min;
+    }
+
     /**
      * Pure decision function: examine everything pending for {@code lead} — reply targets and
      * tickets alike — and return what the loop should do.
@@ -155,9 +202,14 @@ public final class ReplyPushLoop {
      * {@link Action#INJECT}. Only when neither source has eligible work does the loop
      * {@link Action#STOP}.
      *
+     * <p><strong>CB-598: the counts are per-item, not per-tick.</strong> {@link #tick} no longer
+     * carries these counts forward across scheduled calls — it recomputes them fresh every tick via
+     * {@link #minReplyNudgeCountFor} / {@link #minTicketNudgeCountFor}, so this function itself did
+     * not need to change; only what its caller feeds it did.
+     *
      * @param lead                the lead terminal to nudge
-     * @param replyReminderCount  how many nudges have covered pending reply work for this lead
-     * @param ticketReminderCount how many nudges have covered pending ticket work for this lead
+     * @param replyReminderCount  the lowest nudge count among reply targets pending for this lead
+     * @param ticketReminderCount the lowest nudge count among tickets pending for this lead
      * @return the action the caller should take
      */
     Action decide(String lead, int replyReminderCount, int ticketReminderCount) {
@@ -204,7 +256,8 @@ public final class ReplyPushLoop {
             log.debug("push: no lead is known to be waiting on {}, skipping reminder", target);
             return;
         }
-        pendingReplies.put(target, lead.get());
+        pendingReplies.compute(target, (t, existing) ->
+                new ReplyEntry(lead.get(), existing == null ? 0 : existing.nudgeCount()));
         startOrCoalesce(lead.get());
     }
 
@@ -232,7 +285,8 @@ public final class ReplyPushLoop {
                     ticket, target);
             return;
         }
-        pendingTickets.put(ticket, new PendingTicket(ticket, lead.get(), failed));
+        pendingTickets.compute(ticket, (id, existing) ->
+                new PendingTicket(ticket, lead.get(), failed, existing == null ? 0 : existing.nudgeCount()));
         startOrCoalesce(lead.get());
     }
 
@@ -255,28 +309,37 @@ public final class ReplyPushLoop {
             return;
         }
         log.debug("push: starting reminder loop for lead {}", lead);
-        scheduleNext(lead, 0, 0);
+        scheduleNext(lead);
     }
 
-    /** Execute one loop tick — called on the scheduler thread. */
-    private void tick(String lead, int replyReminderCount, int ticketReminderCount) {
+    /**
+     * Execute one loop tick — called on the scheduler thread (or directly by a test; package-private
+     * for the same reason as {@link #stopOrRestart}).
+     *
+     * <p><strong>CB-598.</strong> The reminder counts fed into {@link #decide} are recomputed fresh
+     * every tick from what is actually pending right now ({@link #minReplyNudgeCountFor} /
+     * {@link #minTicketNudgeCountFor}), rather than carried forward as running counters across
+     * scheduled calls. A counter carried forward has no memory of which item it was counting for:
+     * a target or ticket that joined mid-backoff — after the previous tick fired but before this one
+     * did — is already sitting in {@code repliesBefore} / {@code ticketsBefore} below by the time this
+     * tick takes its snapshot, indistinguishable at that point from backlog the cap is meant to
+     * silence. Recomputing from the per-item counts fixes that: a newly-joined item's own count is
+     * still 0, so it keeps its source eligible regardless of how depleted an older, still-undrained
+     * item's count is.
+     */
+    void tick(String lead) {
         Set<String> repliesBefore = pendingReplyTargetsFor(lead);
         Set<String> ticketsBefore = pendingTicketIdsFor(lead);
+        int replyReminderCount = minReplyNudgeCountFor(lead);
+        int ticketReminderCount = minTicketNudgeCountFor(lead);
         var action = decide(lead, replyReminderCount, ticketReminderCount);
         switch (action) {
             case INJECT -> {
                 injectNudge(lead, replyReminderCount, ticketReminderCount);
-                // Only the source(s) actually eligible this tick spend a unit of their own budget —
-                // an exhausted source riding along in the combined message (still pending, still
-                // named) does not get charged again; its count stays put until it drains.
-                boolean replyEligible = !repliesBefore.isEmpty() && replyReminderCount < maxReminders;
-                boolean ticketEligible = !ticketsBefore.isEmpty() && ticketReminderCount < maxReminders;
-                scheduleNext(lead,
-                        replyEligible ? replyReminderCount + 1 : replyReminderCount,
-                        ticketEligible ? ticketReminderCount + 1 : ticketReminderCount);
+                scheduleNext(lead);
             }
             // Re-check after the configured backoff; the lead may become injectable soon.
-            case WAIT_BUSY -> scheduleNext(lead, replyReminderCount, ticketReminderCount);
+            case WAIT_BUSY -> scheduleNext(lead);
             case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore);
         }
     }
@@ -317,7 +380,7 @@ public final class ReplyPushLoop {
                 || pendingTicketIdsFor(lead).stream().anyMatch(t -> !ticketsBefore.contains(t));
         if (racedIn && activeLeads.putIfAbsent(lead, Boolean.TRUE) == null) {
             log.debug("push: new work for lead {} raced the reminder loop's stop — restarting", lead);
-            scheduleNext(lead, 0, 0);
+            scheduleNext(lead);
             return;
         }
         log.debug("push: reminder loop ended for lead {}", lead);
@@ -344,11 +407,28 @@ public final class ReplyPushLoop {
             log.warn("push: failed to nudge lead {} (reply {}/{}, ticket {}/{}): {}",
                     lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders, e.toString());
         }
+        // Bump every item actually named in this nudge, not just whatever the shared source-level
+        // eligibility used to gate (CB-598) — each item's own count is what the next tick's
+        // minReplyNudgeCountFor / minTicketNudgeCountFor will read. An item already at or over the
+        // cap keeps riding along in the text (still pending, still named) but its extra bumps here
+        // are inert: decide() already treats it as ineligible once its count reaches maxReminders.
+        bumpNudgeCounts(replyTargets, tickets);
+    }
+
+    /** Record that every one of these items was just named in a sent (or attempted) nudge. */
+    private void bumpNudgeCounts(Set<String> replyTargets, List<PendingTicket> tickets) {
+        for (String target : replyTargets) {
+            pendingReplies.computeIfPresent(target, (t, e) -> new ReplyEntry(e.lead(), e.nudgeCount() + 1));
+        }
+        for (PendingTicket ticket : tickets) {
+            pendingTickets.computeIfPresent(ticket.ticket(),
+                    (id, e) -> new PendingTicket(e.ticket(), e.lead(), e.failed(), e.nudgeCount() + 1));
+        }
     }
 
     /** Schedule the next tick on the scheduler thread pool. */
-    private void scheduleNext(String lead, int nextReplyReminderCount, int nextTicketReminderCount) {
-        scheduler.schedule(() -> tick(lead, nextReplyReminderCount, nextTicketReminderCount),
+    private void scheduleNext(String lead) {
+        scheduler.schedule(() -> tick(lead),
                 backoffMs, TimeUnit.MILLISECONDS);
     }
 

bridged/src/main/java/dev/ltms/bridged/rest/BridgedApp.java

@@ -13,7 +13,6 @@ import dev.ltms.bridged.herdr.HerdrClient;
 import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.inject.MemberPresence;
 import dev.ltms.bridged.peer.PeerUnreachableException;
-import dev.ltms.bridged.placement.PlacementException;
 import dev.ltms.bridged.msg.MessageService;
 import dev.ltms.bridged.session.SessionManager;
 import dev.ltms.bridged.peer.MemberRole;
@@ -293,11 +292,6 @@ public final class BridgedApp {
             ctx.status(201).json(view(member));
         } catch (GuardException e) {
             ctx.status(403).json(Map.of("error", "subscription_boundary", "detail", e.getMessage()));
-        } catch (PlacementException e) {
-            // CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — a benign,
-            // likely-transient refusal, distinct from "profile does not exist" below. 503: the
-            // request was valid and will likely succeed later.
-            ctx.status(503).json(Map.of("error", "no_capacity", "detail", e.getMessage()));
         } catch (IllegalArgumentException e) {
             ctx.status(400).json(Map.of("error", "unknown_profile", "detail", e.getMessage()));
         } catch (PeerUnreachableException e) {

bridged/src/test/java/dev/ltms/bridged/config/BridgedConfigTest.java

@@ -10,7 +10,6 @@ import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.regex.Pattern;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -1190,79 +1189,6 @@ class BridgedConfigTest {
         assertEquals(5, cfg.leadHeartbeat().quietNudgeCap());
     }
 
-    /**
-     * A top-level key {@code BridgedConfig} reads but that appears nowhere in
-     * {@code bridged.example.yaml} — live or commented — is invisible drift: {@code bridged.yaml}
-     * is gitignored, so the example is the ONLY committed description of the config schema, and
-     * neither {@link #shippedExampleConfigParses} (example → code: does the example still parse)
-     * nor {@link #everyOptionalKnobDocumentedInTheExampleBinds} (a hand-maintained list of keys
-     * that must bind) can catch a brand-new key nobody added to either.
-     *
-     * <p>This test compares the OTHER direction: every key in {@link BridgedConfig#KNOWN_TOP_LEVEL_KEYS}
-     * (the parser's own accepted set, which backs the unknown-key WARN) must appear as a top-level
-     * key in the example text, live or commented-out — see {@link #topLevelKeyDocumented}.
-     */
-    @Test
-    void everyKnownTopLevelKeyIsDocumentedInTheExample() throws Exception {
-        Path example = Path.of("bridged.example.yaml");
-        assertTrue(Files.exists(example), "bridged.example.yaml must ship next to the pom");
-        String text = Files.readString(example);
-
-        List<String> undocumented = BridgedConfig.KNOWN_TOP_LEVEL_KEYS.stream()
-                .filter(key -> !topLevelKeyDocumented(text, key))
-                .sorted()
-                .toList();
-
-        assertTrue(undocumented.isEmpty(), () -> "key(s) " + undocumented
-                + " are read by BridgedConfig but appear nowhere in bridged.example.yaml — "
-                + "document each one there, commented out if optional. bridged.yaml is "
-                + "gitignored, so this file is the only committed description of the config "
-                + "schema an operator or a worker can see.");
-    }
-
-    /**
-     * Most of {@code bridged.example.yaml} is deliberately commented out — optional sections are
-     * documented as commented blocks so the shipped file stays a working minimal config. A key
-     * documented ONLY as a comment must still count as documented; parsing the file as YAML and
-     * reading its live key set (as an earlier attempt at this guard did) gets this wrong, because
-     * every commented section then looks entirely absent.
-     */
-    @Test
-    void commentedOnlyTopLevelKeyCountsAsDocumented() {
-        String yaml = """
-                bind:
-                  port: 8765
-                # broker:
-                #   uri: amqp://guest:guest@127.0.0.1:5672
-                """;
-        assertTrue(topLevelKeyDocumented(yaml, "broker"),
-                "a key documented only inside a commented-out block must still count as documented");
-    }
-
-    /** A key that appears in neither a live nor a commented top-level line must NOT count. */
-    @Test
-    void absentTopLevelKeyIsNotDocumented() {
-        String yaml = """
-                bind:
-                  port: 8765
-                """;
-        assertFalse(topLevelKeyDocumented(yaml, "broker"),
-                "a key mentioned nowhere in the example must not be reported as documented");
-    }
-
-    /**
-     * True when {@code key} appears as a top-level YAML key in {@code yaml} — either live
-     * ({@code key:} at column 0) or commented out ({@code # key:}, also at column 0, with only
-     * whitespace between the {@code #} and the key). Anchoring on column 0 is what keeps this a
-     * top-level check: an indented occurrence (a nested field, or prose inside a comment that
-     * happens to end in a colon) never matches, because {@code ^} requires the key's own first
-     * character — or the sole leading {@code #} — to sit at the very start of the line.
-     */
-    private static boolean topLevelKeyDocumented(String yaml, String key) {
-        Pattern p = Pattern.compile("(?m)^(?:#\\s*)?" + Pattern.quote(key) + ":");
-        return p.matcher(yaml).find();
-    }
-
     @Test
     void placementDefaultsToFixedForExistingConfigs(@TempDir Path dir) throws Exception {
         Path f = dir.resolve("no-placement.yaml");

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

@@ -16,15 +16,12 @@ import dev.ltms.bridged.inject.MemberPresence;
 import dev.ltms.bridged.session.MemberSession;
 import dev.ltms.bridged.session.WorktreeRequest;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
-import dev.ltms.bridged.member.CompositePeerLauncher;
 import dev.ltms.bridged.placement.BackendQuarantine;
-import dev.ltms.bridged.placement.PlacementPolicies;
 import io.modelcontextprotocol.spec.McpSchema;
 import dev.ltms.bridged.msg.InMemoryReplyInbox;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -426,35 +423,6 @@ class BridgeMcpTest {
         assertTrue(textOf(res).contains("unknown worker profile"), textOf(res));
     }
 
-    /**
-     * CB-599: a profile at its {@code maxLoad} cap must surface a readable reason on the MCP
-     * surface too, not merely flip {@code isError} with an opaque or absent message.
-     */
-    @Test
-    void spawnAtMaxLoadSurfacesTheCapacityReason() {
-        FakeHerdr h = new FakeHerdr();
-        BridgedConfig.Profile wcfg = new BridgedConfig.Profile(
-                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN", null,
-                "tab", "bridged-workers", "worker: {profile} #{n}", null,
-                null, null, null, null, null, null, null, 0, null, null, null);
-        Map<String, BridgedConfig.Profile> profiles = Map.of(wcfg.profile(), wcfg);
-        ClaudeCodeLauncher delegate = new ClaudeCodeLauncher(
-                new AgentControl(h), new WorkspaceControl(h), new SubscriptionGuard(Set.of("gx00.gw")),
-                profiles, wcfg.profile(), k -> "BRIDGED_WORKER_TOKEN".equals(k) ? "tok" : null);
-        CompositePeerLauncher composite = new CompositePeerLauncher(
-                List.of(delegate), wcfg.profile(), profiles, PlacementPolicies.fixed(), _ -> 0);
-        SessionManager sm = new SessionManager(composite);
-
-        McpSchema.CallToolResult res = BridgeMcp.spawn(sm, "ltms-local");
-
-        assertTrue(res.isError());
-        String text = textOf(res);
-        assertTrue(text.contains("no capacity"), "surfaces a capacity reason, not a bare error: " + text);
-        assertTrue(text.contains("ltms-local"), "names the profile: " + text);
-        assertTrue(text.contains("maxLoad"), "explains the refusal: " + text);
-        assertFalse(h.called("agent.start"), "at cap, the spawn is refused before any herdr call");
-    }
-
     @Test
     void spawnPassesTheRequestedCwdToTheWorker() {
         FakeHerdr h = new FakeHerdr();

bridged/src/test/java/dev/ltms/bridged/msg/ReplyPushLoopTest.java

@@ -42,6 +42,7 @@ class ReplyPushLoopTest {
 
     private static final String PRIMARY = "term_primary";
     private static final String WORKER = "term_worker";
+    private static final String WORKER2 = "term_worker2";
     private static final ObjectMapper MAPPER = new ObjectMapper();
 
     private PrimaryRegistry registry;
@@ -579,6 +580,83 @@ class ReplyPushLoopTest {
                 "both the reply and the ticket source are at their own cap — must still stop");
     }
 
+    // --- CB-598: work arriving during a backoff must not read as stale backlog -------------------
+
+    @Test
+    void aTargetArrivingDuringTheBackoffGetsNudgedDespiteAnAlreadyCappedSibling() {
+        // The bug: reminder counts used to be a single counter per lead per source, carried
+        // forward across scheduled ticks (scheduleNext(lead, count + 1, ...)) rather than tracked
+        // per pending item. WORKER gets nudged once here, which — with cap=1 — exhausts the
+        // shared reply-source counter for this lead. WORKER2 then queues a reply for the SAME
+        // lead "during the backoff": while the schedule from WORKER's tick is still active, before
+        // the next tick's own start-of-tick snapshot runs. At that next tick, the OLD code passed
+        // the already-exhausted shared counter into decide() regardless of WORKER2 never having
+        // been named in any nudge, and — because WORKER2 was already present in that tick's
+        // "before" snapshot — stopOrRestart's race check (proven correct on its own elsewhere in
+        // this file) does not save it either: it looks like ordinary stale backlog, not a race.
+        // WORKER2 was then stranded forever with no live schedule and no nudge ever naming it.
+        //
+        // tick() is driven directly (package-private, same reasoning as stopOrRestart being
+        // directly testable) so the exact interleaving is deterministic instead of racing the
+        // scheduler thread over a real ~15s backoff.
+        //
+        // Before the fix, this test fails on the second assertEquals: rec.sendCount() stays at 1
+        // (decide() returns STOP on the second tick(), so injectNudge is never called a second
+        // time) and the "must still get one" assertion never even runs.
+        int cap = 1;
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        inbox.own(WORKER2);
+        inbox.publish(WORKER, "m1", "hello");
+        var loop = loop(cap, 100_000); // huge backoff — nothing fires on its own; we drive tick()
+
+        loop.onReplyQueued(WORKER);
+        loop.tick(PRIMARY); // first tick: nudges WORKER alone; WORKER's own count reaches the cap
+        assertEquals(1, rec.sendCount(), "the first tick should nudge about WORKER");
+
+        // WORKER2 "arrives during the backoff": queued for the same lead while the schedule from
+        // the tick above is still active (activeLeads still holds PRIMARY), before the next tick
+        // (simulated below) takes its own start-of-tick snapshot.
+        inbox.publish(WORKER2, "m2", "hello2");
+        loop.onReplyQueued(WORKER2);
+
+        loop.tick(PRIMARY); // the tick that would fire once that backoff elapsed
+
+        assertEquals(2, rec.sendCount(),
+                "WORKER2 was never named in any nudge yet and must still get one, even though "
+                        + "WORKER's own reminder count is already at the cap");
+        String secondNudge = rec.sentParams().get(1).getValue().toString();
+        assertTrue(secondNudge.contains(WORKER2), "the never-named target must be named: " + secondNudge);
+
+        // Criterion #3: isActive() must reflect that this lead still had a live nudge to give —
+        // the second tick took the INJECT branch, so the schedule stayed live rather than being
+        // torn down under WORKER2.
+        assertTrue(loop.isActive(), "the schedule must stay active after nudging the fresh target");
+    }
+
+    @Test
+    void aTicketArrivingDuringTheBackoffGetsNudgedDespiteAnAlreadyCappedSibling() {
+        // Mirrors the reply-side test above for the ticket source.
+        int cap = 1;
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        var loop = loop(cap, 100_000);
+
+        loop.onTicketTerminal("task-1", WORKER, false);
+        loop.tick(PRIMARY); // first tick: nudges task-1 alone; its count reaches the cap
+        assertEquals(1, rec.sendCount(), "the first tick should nudge about task-1");
+
+        loop.onTicketTerminal("task-2", WORKER, false); // arrives during the backoff, same lead
+        loop.tick(PRIMARY);
+
+        assertEquals(2, rec.sendCount(),
+                "task-2 was never named in any nudge yet and must still get one, even though "
+                        + "task-1's reminder count is already at the cap");
+        String secondNudge = rec.sentParams().get(1).getValue().toString();
+        assertTrue(secondNudge.contains("task-2"), "the never-named ticket must be named: " + secondNudge);
+        assertTrue(loop.isActive(), "the schedule must stay active after nudging the fresh ticket");
+    }
+
     // --- metrics (CB-512) ----------------------------------------------------------------------
 
     @Test

bridged/src/test/java/dev/ltms/bridged/rest/BridgedAppTest.java

@@ -18,8 +18,6 @@ import dev.ltms.bridged.session.GitWorktrees;
 import dev.ltms.bridged.session.SessionManager;
 import dev.ltms.bridged.session.Worktrees;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
-import dev.ltms.bridged.member.CompositePeerLauncher;
-import dev.ltms.bridged.placement.PlacementPolicies;
 import io.javalin.Javalin;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
@@ -242,43 +240,6 @@ class BridgedAppTest {
         assertFalse(herdr.called("agent.start"), "an unknown profile must not spawn anything");
     }
 
-    /**
-     * CB-599: a profile at its {@code maxLoad} cap must not surface as a bare 500 — the caller
-     * needs a structured, readable reason, distinct from "unknown_profile".
-     */
-    @Test
-    void spawnAtMaxLoadIs503WithTheCapacityReasonNotABare500() throws Exception {
-        FakeHerdr herdr = new FakeHerdr();
-        BridgedConfig.Profile wcfg = new BridgedConfig.Profile(
-                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN", null,
-                "tab", "bridged-workers", "worker: {profile} #{n}", null,
-                null, null, null, null, null, null, null, 0, null, null, null);
-        Map<String, BridgedConfig.Profile> profiles = Map.of(wcfg.profile(), wcfg);
-        ClaudeCodeLauncher delegate = new ClaudeCodeLauncher(
-                new AgentControl(herdr), new WorkspaceControl(herdr), new SubscriptionGuard(Set.of("gx00.gw")),
-                profiles, wcfg.profile(), k -> "BRIDGED_WORKER_TOKEN".equals(k) ? "tok-abc" : null);
-        CompositePeerLauncher workers = new CompositePeerLauncher(
-                List.of(delegate), wcfg.profile(), profiles, PlacementPolicies.fixed(), _ -> 0);
-        SessionManager sessions = new SessionManager(workers, new GitWorktrees());
-        this.presence = sessions.asPresence();
-        Injector injector = new Injector(new AgentControl(herdr));
-        InMemoryReplyInbox inbox = new InMemoryReplyInbox();
-        sessions.onAcquire(inbox::own);
-        MessageService messages = new MessageService(new AgentControl(herdr), injector, new Rendezvous(), inbox);
-        app = new BridgedApp(herdr, workers, sessions, messages, this.presence, null).build().start("127.0.0.1", 0);
-        int port = app.port();
-
-        HttpResponse<String> res = req(port, "POST", "/members?profile=ltms-local");
-
-        assertEquals(503, res.statusCode(), res.body());
-        JsonNode body = mapper.readTree(res.body());
-        assertEquals("no_capacity", body.get("error").asText());
-        String detail = body.get("detail").asText();
-        assertTrue(detail.contains("ltms-local"), "detail names the profile: " + detail);
-        assertTrue(detail.contains("maxLoad"), "detail explains the refusal: " + detail);
-        assertFalse(herdr.called("agent.start"), "at cap, the spawn is refused before any herdr call");
-    }
-
     @Test
     void spawnWorkerReusesExistingWorkerSpace() throws Exception {
         // A space labelled "bridged-workers" already exists → no second workspace.create.