CB-597: fix inaccuracies the example config already had, none actually missing

Audited bridged.example.yaml against BridgedConfig's KNOWN_TOP_LEVEL_KEYS and
found every top-level key already documented (broker, health, lifecycle,
configReload, quarantineCooldownSeconds, fleet.leaders/architects/reviewers
included) — CB-573/CB-566/CB-559/CB-579/CB-527/528 each updated the example
alongside their feature. What was actually wrong:

- health.workingSuspectAfterSeconds/paneProbeIntervalSeconds claimed enforced
  minimums (300/60) that don't exist in code — only intervalSeconds is
  clamped (floor 15); the other two are parsed but never read anywhere.
- notifications.mode: webhook was undocumented as only flipping the
  healthCoverage label bridge_list reports — no webhook is ever sent.
- lifecycle.clearAfterTurn was missing entirely.
- the HOT bullet under configReload claimed the whole fleet: block reloads
  live, but ConfigRef's own javadoc carves out fleet.leaders as needing a
  restart with no deferred-list warning — added that exception.
- fleet.leaders' demotion consequence (unmatched tab -> silent WORKER
  demotion, no startup error) is now stated inline next to the block, not
  just implied by the multi-lead rationale higher up.

bridged/bridged.example.yaml

@@ -88,15 +88,28 @@ bind:
 #   backoffMs: 60000
 #   quietNudgeCap: 3
 
-# 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.
+# 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".
 # health:
 #   enabled: true
-#   intervalSeconds: 30              # minimum 15
-#   workingSuspectAfterSeconds: 600  # minimum 300
-#   paneProbeIntervalSeconds: 60     # minimum 60
+#   intervalSeconds: 30
+#   workingSuspectAfterSeconds: 600
+#   paneProbeIntervalSeconds: 60
 #   notifications:
-#     mode: disabled                 # disabled (default) or webhook
+#     mode: disabled
 
 # herdr Unix socket. Omit to use the client default
 # (${HERDR_SOCKET_PATH:-~/.config/herdr/herdr.sock}).
@@ -286,6 +299,11 @@ 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
@@ -368,6 +386,12 @@ 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
@@ -423,10 +447,15 @@ 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).
@@ -434,10 +463,14 @@ guard:
 # on a durable per-target queue (agent.<target>.inbox) and survive a restart — the broker
 # redelivers anything the primary had not yet drained. Production default is LavinMQ; a stock
 # RabbitMQ speaks the same AMQP 0-9-1, so it is a URI-only swap.
-#   uri → AMQP connection URI. No trailing slash ⇒ the default vhost "/"; an empty path ("/")
-#         is vhost "" and will NOT connect. Encode a named vhost as .../%2Fmyvhost.
+#   uri      → AMQP connection URI. No trailing slash ⇒ the default vhost "/"; an empty path ("/")
+#              is vhost "" and will NOT connect. Encode a named vhost as .../%2Fmyvhost.
+#   prefetch → CB-527: consumer basicQos, capping how many unacked messages the inbox holds
+#              in-heap per owned target (the rest sits on the broker's durable queue instead of
+#              growing the JVM heap). Default 32 when omitted.
 # broker:
 #   uri: amqp://guest:guest@127.0.0.1:5672
+#   prefetch: 32
 
 # Active push-to-primary (CB-307 Stage 3). When a worker reply lands with no open bridge_send,
 # the ReplyPushLoop injects a *drain nudge* (never the payload) into the primary's own herdr

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

@@ -348,8 +348,9 @@ public final class Bridged {
         // connection, so keep the reference to close it in the ordered shutdown hook.
         final ReplyInbox replyInbox;
         if (cfg.broker() != null && cfg.broker().isConfigured()) {
-            replyInbox = AmqpReplyInbox.open(cfg.broker().uri());
-            log.info("reply inbox: AMQP broker (durable) at {}", cfg.broker().uri());
+            replyInbox = AmqpReplyInbox.open(cfg.broker().uri(), cfg.broker().prefetchOrDefault());
+            log.info("reply inbox: AMQP broker (durable) at {} (prefetch={})",
+                    cfg.broker().uri(), cfg.broker().prefetchOrDefault());
         } else {
             replyInbox = new InMemoryReplyInbox();
             log.info("reply inbox: in-memory (soft-state)");

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

@@ -5,6 +5,7 @@ import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonToken;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import dev.ltms.bridged.msg.AmqpReplyInbox;
 import dev.ltms.bridged.peer.MemberRole;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -532,16 +533,24 @@ public record BridgedConfig(
      * stays soft-state. Production default is LavinMQ; a stock RabbitMQ speaks the same AMQP 0-9-1
      * and is a URI-only swap.
      *
-     * @param uri AMQP connection URI, e.g. {@code amqp://guest:guest@127.0.0.1:5672/}. Blank/{@code null}
-     *            ⇒ the broker block is treated as absent (in-memory adapter).
+     * @param uri      AMQP connection URI, e.g. {@code amqp://guest:guest@127.0.0.1:5672/}. Blank/
+     *                 {@code null} ⇒ the broker block is treated as absent (in-memory adapter).
+     * @param prefetch CB-527: the consumer's {@code basicQos} prefetch count, bounding how many
+     *                 unacked messages the AMQP inbox holds in-heap per owned target. {@code null}/
+     *                 non-positive ⇒ {@link AmqpReplyInbox#DEFAULT_PREFETCH}.
      */
     @JsonIgnoreProperties(ignoreUnknown = true)
-    public record Broker(String uri) {
+    public record Broker(String uri, Integer prefetch) {
 
         /** True when a usable broker URI is configured (an empty block does not enable AMQP). */
         public boolean isConfigured() {
             return uri != null && !uri.isBlank();
         }
+
+        /** The prefetch to use, defaulting to {@link AmqpReplyInbox#DEFAULT_PREFETCH} when unset. */
+        public int prefetchOrDefault() {
+            return (prefetch != null && prefetch > 0) ? prefetch : AmqpReplyInbox.DEFAULT_PREFETCH;
+        }
     }
 
     /**

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

@@ -7,6 +7,7 @@ import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.DeliverCallback;
 import com.rabbitmq.client.Recoverable;
 import com.rabbitmq.client.RecoveryListener;
+import com.rabbitmq.client.Return;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -14,7 +15,13 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.NavigableMap;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 /**
  * AMQP-backed {@link ReplyInbox} (CB-307 Stage 2): genuine cross-restart durability behind the same
@@ -29,6 +36,27 @@ import java.util.concurrent.ConcurrentHashMap;
  * leaves them on the broker — it redelivers on reconnect. That is the durability the in-memory
  * adapter cannot give, with the port contract preserved.
  *
+ * <p><strong>Prefetch bounds the held backlog (CB-527).</strong> The consumer channel calls
+ * {@code basicQos} with a configurable prefetch count ({@link #DEFAULT_PREFETCH} unless the caller
+ * passes another value to {@link #open(String, int)}) before starting any consumer. Without a bound,
+ * the broker pushes its entire queue into {@link #held} the instant a target is {@link #own owned},
+ * so an undrained primary grows the JVM heap without limit and any queue-level control
+ * ({@code x-max-length}, per-message TTL) never fires because the queue never actually holds a
+ * backlog. Prefetch keeps the backlog where it is visible — on the broker — until the owner drains it.
+ *
+ * <p><strong>Publishes require a confirmed, routable delivery (CB-528).</strong> {@link #publish}
+ * runs on a channel separate from the consume/ack channel ({@link #channel}), so a slow or blocked
+ * publish confirm can never hold {@link #channelLock} and stall an ack — the ack path never waits on
+ * a publish confirm. That publish channel is in publisher-confirm mode and every publish sets the
+ * {@code mandatory} flag, so an unroutable publish (queue not declared, e.g. the owner never called
+ * {@link #own}) is returned by the broker instead of silently dropped. The broker sends the
+ * <em>return</em> for an unroutable message before the <em>confirm</em> that covers it — the ack/nack
+ * callback checks the returned-set at confirm time rather than assuming an ack means routed — so
+ * "confirmed" here means "durably queued", not merely "accepted by the broker". A returned or nacked
+ * (or un-confirmed within the timeout) publish surfaces as an {@link IllegalStateException} on the
+ * caller's thread; the caller — {@link MessageService#reply} — must not report success for a
+ * black-holed reply.
+ *
  * <p><strong>Ownership is explicit.</strong> {@link #own} declares the queue and starts the consumer;
  * {@link #release} cancels it. {@link #publish} sends to the queue but does <em>not</em> imply ownership
  * and does not attach a consumer. This split is required by CB-308 federation, where one gateway may
@@ -54,6 +82,12 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     private static final String QUEUE_PREFIX = "agent.";
     private static final String QUEUE_SUFFIX = ".inbox";
 
+    /** CB-527: the prefetch used when a caller does not pass an explicit value to {@link #open(String, int)}. */
+    public static final int DEFAULT_PREFETCH = 32;
+
+    /** How long {@link #publish} waits for its publisher confirm before failing the call (CB-528). */
+    private static final long CONFIRM_TIMEOUT_MS = 10_000L;
+
     private final Connection connection;
     private final Channel channel;
     /** All channel operations (publish/declare/ack/cancel) serialize on this — a Channel is not thread-safe. */
@@ -63,39 +97,87 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     /** Targets whose queue is declared and consumer is running, mapped to their broker consumer tag. */
     private final ConcurrentHashMap<String, String> consumerTags = new ConcurrentHashMap<>();
 
+    /**
+     * CB-528: a dedicated channel for {@link #publish}, kept separate from {@link #channel} (consume
+     * + ack) so a publish confirm round trip never blocks under {@link #channelLock} and stalls an ack.
+     */
+    private final Channel publishChannel;
+    private final Object publishChannelLock = new Object();
+    /** In-flight publishes awaiting their confirm, keyed by the publish channel's sequence number. */
+    private final ConcurrentSkipListMap<Long, Pending> pendingBySeq = new ConcurrentSkipListMap<>();
+    /**
+     * The same in-flight publishes, keyed by {@code msgId} — a broker {@code Return} carries no delivery
+     * tag. Assumes {@code msgId} is unique per in-flight publish: a second {@link #publish} for a
+     * {@code msgId} still awaiting its confirm would overwrite this entry and misdirect
+     * {@link #onReturn}'s lookup. Not reachable today — {@code MessageService.reply} generates a fresh
+     * {@code UUID} per call — so no guard is added for it.
+     */
+    private final ConcurrentHashMap<String, Pending> pendingByMsgId = new ConcurrentHashMap<>();
+
     /** A message pulled off the broker but not yet acked: its delivery-tag plus the port payload. */
     private record Held(long deliveryTag, InboxMessage message) {}
 
-    /** Connect to {@code uri} (e.g. {@code amqp://guest:guest@127.0.0.1:5672/}) and open the inbox. */
+    /** A publish awaiting its confirm; {@link #returned} records whether the broker already returned it. */
+    private static final class Pending {
+        final String msgId;
+        final CompletableFuture<Void> confirmed = new CompletableFuture<>();
+        volatile boolean returned;
+
+        Pending(String msgId) {
+            this.msgId = msgId;
+        }
+    }
+
+    /** Connect to {@code uri} (e.g. {@code amqp://guest:guest@127.0.0.1:5672/}) with {@link #DEFAULT_PREFETCH}. */
     public static AmqpReplyInbox open(String uri) {
+        return open(uri, DEFAULT_PREFETCH);
+    }
+
+    /** As {@link #open(String)}, with an explicit consumer prefetch (CB-527: caps the held backlog per target). */
+    public static AmqpReplyInbox open(String uri, int prefetch) {
         try {
             ConnectionFactory factory = new ConnectionFactory();
             factory.setUri(uri);
             // Self-heal transient blips; topology recovery re-declares queues and re-attaches consumers.
             factory.setAutomaticRecoveryEnabled(true);
             factory.setTopologyRecoveryEnabled(true);
-            return new AmqpReplyInbox(factory.newConnection("bridged-reply-inbox"));
+            return new AmqpReplyInbox(factory.newConnection("bridged-reply-inbox"), prefetch);
         } catch (Exception e) {
             throw new IllegalStateException("cannot connect to AMQP broker at " + uri, e);
         }
     }
 
-    /** Wrap an already-open connection (injection seam for the contract test). */
+    /** Wrap an already-open connection with {@link #DEFAULT_PREFETCH} (injection seam for the contract test). */
     AmqpReplyInbox(Connection connection) {
+        this(connection, DEFAULT_PREFETCH);
+    }
+
+    /** As above, with an explicit prefetch (injection seam for the contract test). */
+    AmqpReplyInbox(Connection connection, int prefetch) {
         this.connection = connection;
         try {
             this.channel = connection.createChannel();
+            // CB-527: bound the held backlog per owned target — must be set before any own()/basicConsume.
+            this.channel.basicQos(prefetch);
+            this.publishChannel = connection.createChannel();
+            this.publishChannel.confirmSelect();
+            this.publishChannel.addReturnListener(this::onReturn);
+            this.publishChannel.addConfirmListener(this::onAck, this::onNack);
         } catch (IOException e) {
             throw new IllegalStateException("cannot open AMQP channel", e);
         }
         // On automatic recovery the broker redelivers unacked messages with FRESH delivery-tags; the
         // tags we were holding are now stale. Drop the held snapshot so the re-attached consumer
-        // repopulates it with valid tags (dedup by msgId still prevents any double-queue).
+        // repopulates it with valid tags (dedup by msgId still prevents any double-queue). Any publish
+        // confirm still in flight when the connection dropped is equally stale — its sequence number
+        // meant nothing on the old channel and means nothing on the recovered one, so fail it now
+        // rather than let it silently ride out CONFIRM_TIMEOUT_MS.
         if (connection instanceof Recoverable recoverable) {
             recoverable.addRecoveryListener(new RecoveryListener() {
                 @Override
                 public void handleRecovery(Recoverable recoverable) {
                     held.clear();
+                    failPendingPublishesOnRecovery();
                     log.info("AMQP connection recovered; cleared held replies for fresh redelivery");
                 }
 
@@ -141,6 +223,12 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
         }
     }
 
+    /**
+     * Publish {@code content} and block until the broker's publisher confirm for it lands (CB-528).
+     * Throws {@link IllegalStateException} if the message is returned as unroutable, nacked, or not
+     * confirmed within {@link #CONFIRM_TIMEOUT_MS} — the caller must treat that as a failed publish,
+     * not a lost-and-forgotten one.
+     */
     @Override
     public void publish(String target, String msgId, String content) {
         AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
@@ -148,12 +236,35 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
                 .deliveryMode(2)                   // persistent — survives a broker restart
                 .contentType("text/plain")
                 .build();
-        try {
-            synchronized (channelLock) {
-                channel.basicPublish("", queueName(target), props, content.getBytes(StandardCharsets.UTF_8));
+        Pending pending = new Pending(msgId);
+        long seq;
+        synchronized (publishChannelLock) {
+            seq = publishChannel.getNextPublishSeqNo();
+            pendingBySeq.put(seq, pending);
+            pendingByMsgId.put(msgId, pending);
+            try {
+                publishChannel.basicPublish("", queueName(target), true, props,
+                        content.getBytes(StandardCharsets.UTF_8));
+            } catch (IOException e) {
+                pendingBySeq.remove(seq, pending);
+                pendingByMsgId.remove(msgId, pending);
+                throw new IllegalStateException("cannot publish reply to " + queueName(target), e);
             }
-        } catch (IOException e) {
-            throw new IllegalStateException("cannot publish reply to " + queueName(target), e);
+        }
+        try {
+            pending.confirmed.get(CONFIRM_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause();
+            throw cause instanceof RuntimeException re ? re : new IllegalStateException(cause);
+        } catch (TimeoutException e) {
+            throw new IllegalStateException("publish confirm for reply " + msgId + " to " + queueName(target)
+                    + " timed out after " + CONFIRM_TIMEOUT_MS + "ms — broker may be unreachable or overloaded", e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("interrupted awaiting publish confirm for " + msgId, e);
+        } finally {
+            pendingBySeq.remove(seq, pending);
+            pendingByMsgId.remove(msgId, pending);
         }
     }
 
@@ -222,17 +333,115 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
         };
     }
 
+    /** Broker return for an unroutable {@code mandatory} publish — arrives BEFORE its confirm (CB-528). */
+    private void onReturn(Return r) {
+        String msgId = r.getProperties() == null ? null : r.getProperties().getMessageId();
+        Pending pending = msgId == null ? null : pendingByMsgId.get(msgId);
+        if (pending != null) {
+            pending.returned = true;
+        } else {
+            log.warn("AMQP return for reply {} (routingKey={}, {} {}) with no matching in-flight publish"
+                    + " — already resolved by a prior confirm", msgId, r.getRoutingKey(), r.getReplyCode(),
+                    r.getReplyText());
+        }
+    }
+
+    private void onAck(long seq, boolean multiple) {
+        resolveConfirm(seq, multiple, true);
+    }
+
+    private void onNack(long seq, boolean multiple) {
+        resolveConfirm(seq, multiple, false);
+    }
+
+    /**
+     * Resolve every pending publish covered by this confirm (a single seq, or — {@code multiple} —
+     * every seq up to and including it). Checks {@link Pending#returned} at confirm time: since the
+     * broker's return for an unroutable message always precedes its confirm, an ack that arrives after
+     * a return means "confirmed but never routed", not "durably queued".
+     */
+    private void resolveConfirm(long seq, boolean multiple, boolean ack) {
+        NavigableMap<Long, Pending> covered = multiple
+                ? pendingBySeq.headMap(seq, true)
+                : pendingBySeq.subMap(seq, true, seq, true);
+        for (var it = covered.entrySet().iterator(); it.hasNext(); ) {
+            Pending pending = it.next().getValue();
+            it.remove();
+            pendingByMsgId.remove(pending.msgId, pending);
+            if (ack && !pending.returned) {
+                pending.confirmed.complete(null);
+            } else if (ack) {
+                pending.confirmed.completeExceptionally(new IllegalStateException(
+                        "reply " + pending.msgId + " was returned as unroutable (queue not declared/owned)"));
+            } else {
+                pending.confirmed.completeExceptionally(new IllegalStateException(
+                        "broker nacked publish of reply " + pending.msgId));
+            }
+        }
+    }
+
+    /**
+     * Fail every publish still awaiting its confirm — their sequence numbers are stale after recovery.
+     * Guarded by {@link #publishChannelLock}, the same lock {@link #publish} holds while it takes its
+     * sequence number and registers its {@link Pending}: without it, a {@link #publish} that starts
+     * after the connection has already recovered (so it publishes — and will be confirmed — on the
+     * <em>new</em> channel) can register between this sweep's iteration and its clear, and this sweep
+     * then fails a publish that actually succeeded. {@link #publish} only holds the lock for the
+     * seq/map-put/{@code basicPublish} — it awaits the confirm outside it — so this sweep can only ever
+     * wait for an in-flight {@code basicPublish} call to return, never for a broker round trip. No
+     * deadlock.
+     *
+     * <p>Package-private (rather than {@code private}) only so the unit test can drive it directly
+     * against a concurrent {@link #publish} without a live broker reconnect.
+     */
+    void failPendingPublishesOnRecovery() {
+        synchronized (publishChannelLock) {
+            for (var it = pendingBySeq.entrySet().iterator(); it.hasNext(); ) {
+                Pending pending = it.next().getValue();
+                it.remove();
+                pendingByMsgId.remove(pending.msgId, pending);
+                pending.confirmed.completeExceptionally(new IllegalStateException(
+                        "AMQP connection recovered mid-publish; confirm status of reply " + pending.msgId
+                                + " is unknown"));
+            }
+        }
+    }
+
+    /**
+     * Fail every publish still awaiting its confirm with a clear, immediate error instead of leaving it
+     * to time out after {@link #CONFIRM_TIMEOUT_MS} once the channels are closed underneath it. Guarded
+     * by {@link #publishChannelLock} for the same reason as {@link #failPendingPublishesOnRecovery}.
+     */
+    private void failPendingPublishesOnClose() {
+        synchronized (publishChannelLock) {
+            for (var it = pendingBySeq.entrySet().iterator(); it.hasNext(); ) {
+                Pending pending = it.next().getValue();
+                it.remove();
+                pendingByMsgId.remove(pending.msgId, pending);
+                pending.confirmed.completeExceptionally(new IllegalStateException(
+                        "AMQP reply inbox closed while publish of reply " + pending.msgId
+                                + " was still awaiting its confirm"));
+            }
+        }
+    }
+
     private static String queueName(String target) {
         return QUEUE_PREFIX + target + QUEUE_SUFFIX;
     }
 
     @Override
     public void close() {
+        failPendingPublishesOnClose();
         try {
             channel.close();
         } catch (Exception e) {
             log.debug("AMQP channel close: {}", e.toString());
         }
+        try {
+            publishChannel.close();
+        } catch (Exception e) {
+            log.debug("AMQP publish channel close: {}", e.toString());
+        }
         try {
             connection.close();
         } catch (Exception e) {

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

@@ -37,10 +37,12 @@ import java.util.stream.Collectors;
  * <p>Each tick examines <em>everything</em> pending for that lead — reply targets whose inbox
  * still holds an unacked message ({@link #pendingReplies}) and tickets not yet collected
  * ({@link #pendingTickets}) — and sends at most one combined nudge per tick
- * ({@link #injectNudge(String, int)}). Work that arrives while the lead is busy is never lost:
- * it is re-read fresh on every tick until the lead is injectable or the shared reminder cap
- * ({@link #maxReminders}) is reached, whichever the durable inbox / pending-ticket set doesn't
- * already answer via {@code STOP}.
+ * ({@link #injectNudge(String, int, int)}). Work that arrives while the lead is busy is never
+ * lost: it is re-read fresh on every tick until the lead is injectable or its own reminder cap
+ * ({@link #maxReminders}) is reached — reply and ticket work each spend from their own budget, so
+ * one source exhausting its cap does not stop nudges about the other (post-CB-590 regression fix;
+ * see {@link #decide}) — whichever the durable inbox / pending-ticket set doesn't already answer
+ * via {@code STOP}.
  */
 public final class ReplyPushLoop {
 
@@ -144,20 +146,32 @@ public final class ReplyPushLoop {
      * Pure decision function: examine everything pending for {@code lead} — reply targets and
      * tickets alike — and return what the loop should do.
      *
-     * @param lead          the lead terminal to nudge
-     * @param reminderCount how many nudges have been sent so far for this lead (shared across
-     *                      both reply and ticket work — CB-590 collapses the reminder cap onto
-     *                      one counter per lead, so alternating sources cannot outrun the bound)
+     * <p><strong>CB-590-fix: one schedule, two budgets.</strong> The single per-lead schedule
+     * (CB-590) still ticks once for both sources, but each source is capped independently —
+     * {@code replyReminderCount} against a reply target still pending, {@code ticketReminderCount}
+     * against a ticket still pending. A busy reply stream that exhausts its own cap must not stop
+     * the loop from nudging about a ticket that still has budget left, and vice versa: either
+     * source being eligible (has pending work AND is under its own cap) is enough for
+     * {@link Action#INJECT}. Only when neither source has eligible work does the loop
+     * {@link Action#STOP}.
+     *
+     * @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
      * @return the action the caller should take
      */
-    Action decide(String lead, int reminderCount) {
-        boolean anyPending = !pendingReplyTargetsFor(lead).isEmpty() || !pendingTicketIdsFor(lead).isEmpty();
-        if (!anyPending) {
+    Action decide(String lead, int replyReminderCount, int ticketReminderCount) {
+        boolean hasReplyWork = !pendingReplyTargetsFor(lead).isEmpty();
+        boolean hasTicketWork = !pendingTicketIdsFor(lead).isEmpty();
+        if (!hasReplyWork && !hasTicketWork) {
             log.debug("push: nothing pending for lead {}, stopping reminder", lead);
             return Action.STOP;
         }
-        if (reminderCount >= maxReminders) {
-            log.debug("push: reminder cap ({}) reached for lead {}, stopping", maxReminders, lead);
+        boolean replyEligible = hasReplyWork && replyReminderCount < maxReminders;
+        boolean ticketEligible = hasTicketWork && ticketReminderCount < maxReminders;
+        if (!replyEligible && !ticketEligible) {
+            log.debug("push: reminder cap ({}) reached for lead {} on every source with pending work, stopping",
+                    maxReminders, lead);
             countNudge("exhausted");
             return Action.STOP;
         }
@@ -241,21 +255,28 @@ public final class ReplyPushLoop {
             return;
         }
         log.debug("push: starting reminder loop for lead {}", lead);
-        scheduleNext(lead, 0);
+        scheduleNext(lead, 0, 0);
     }
 
     /** Execute one loop tick — called on the scheduler thread. */
-    private void tick(String lead, int reminderCount) {
+    private void tick(String lead, int replyReminderCount, int ticketReminderCount) {
         Set<String> repliesBefore = pendingReplyTargetsFor(lead);
         Set<String> ticketsBefore = pendingTicketIdsFor(lead);
-        var action = decide(lead, reminderCount);
+        var action = decide(lead, replyReminderCount, ticketReminderCount);
         switch (action) {
             case INJECT -> {
-                injectNudge(lead, reminderCount);
-                scheduleNext(lead, reminderCount + 1);
+                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);
             }
             // Re-check after the configured backoff; the lead may become injectable soon.
-            case WAIT_BUSY -> scheduleNext(lead, reminderCount);
+            case WAIT_BUSY -> scheduleNext(lead, replyReminderCount, ticketReminderCount);
             case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore);
         }
     }
@@ -296,14 +317,14 @@ 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);
+            scheduleNext(lead, 0, 0);
             return;
         }
         log.debug("push: reminder loop ended for lead {}", lead);
     }
 
     /** Send one combined nudge covering everything currently pending for {@code lead}. */
-    private void injectNudge(String lead, int reminderCount) {
+    private void injectNudge(String lead, int replyReminderCount, int ticketReminderCount) {
         // Re-read rather than threading it down from decide(): a reply can drain, or a ticket be
         // collected (or another arrive), between the decision and the injection.
         Set<String> replyTargets = pendingReplyTargetsFor(lead);
@@ -315,18 +336,20 @@ public final class ReplyPushLoop {
         String nudge = formatNudge(replyTargets, tickets);
         try {
             agents.send(lead, nudge);
-            log.debug("push: nudge {}/{} sent to lead {} ({} reply target(s), {} ticket(s))",
-                    reminderCount + 1, maxReminders, lead, replyTargets.size(), tickets.size());
+            log.debug("push: nudge sent to lead {} (reply {}/{}, ticket {}/{}; {} reply target(s), {} ticket(s))",
+                    lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders,
+                    replyTargets.size(), tickets.size());
             countNudge("delivered");
         } catch (RuntimeException e) {
-            log.warn("push: failed to nudge lead {} (reminder {}/{}): {}",
-                    lead, reminderCount + 1, maxReminders, e.toString());
+            log.warn("push: failed to nudge lead {} (reply {}/{}, ticket {}/{}): {}",
+                    lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders, e.toString());
         }
     }
 
     /** Schedule the next tick on the scheduler thread pool. */
-    private void scheduleNext(String lead, int nextReminderCount) {
-        scheduler.schedule(() -> tick(lead, nextReminderCount), backoffMs, TimeUnit.MILLISECONDS);
+    private void scheduleNext(String lead, int nextReplyReminderCount, int nextTicketReminderCount) {
+        scheduler.schedule(() -> tick(lead, nextReplyReminderCount, nextTicketReminderCount),
+                backoffMs, TimeUnit.MILLISECONDS);
     }
 
     // --- nudge formatting ------------------------------------------------------------------------

bridged/src/test/java/dev/ltms/bridged/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.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
@@ -166,6 +167,88 @@ class AmqpReplyInboxContractTest {
         }
     }
 
+    @Test
+    void prefetchBoundsTheHeldBacklog() throws Exception {
+        String target = "worker-prefetch-" + System.nanoTime();
+        int prefetch = 4;
+        int published = 10;
+        try (AmqpReplyInbox inbox = new AmqpReplyInbox(newConnection(), prefetch);
+             Connection inspect = newConnection()) {
+            inbox.own(target);
+            for (int i = 0; i < published; i++) {
+                inbox.publish(target, "m" + i, "payload " + i);
+            }
+            awaitHeldAtLeast(inbox, target, prefetch);
+
+            long depth;
+            try (Channel ch = inspect.createChannel()) {
+                depth = ch.queueDeclarePassive(queueName(target)).getMessageCount();
+            }
+            assertTrue(depth >= published - prefetch,
+                    "broker should still hold at least " + (published - prefetch)
+                            + " undelivered messages behind a prefetch of " + prefetch + ", saw " + depth);
+
+            drainUntilEmpty(inbox, target, inspect);
+        }
+    }
+
+    @Test
+    void unroutablePublishReportsFailureNotSilentSuccess() throws Exception {
+        String target = "worker-unroutable-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            // Deliberately never own(target): the queue is never declared, so the default-exchange
+            // route to agent.<target>.inbox does not exist and the broker must return the publish.
+            IllegalStateException ex = assertThrows(IllegalStateException.class,
+                    () -> inbox.publish(target, "m1", "nobody home"));
+            assertTrue(ex.getMessage() != null && ex.getMessage().toLowerCase().contains("unroutable"),
+                    "expected an unroutable-publish failure, got: " + ex.getMessage());
+        }
+    }
+
+    @Test
+    void confirmedPublishDeliversNormally() throws Exception {
+        String target = "worker-confirm-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
+            inbox.publish(target, "m1", "confirmed delivery"); // must return normally: routed and confirmed
+
+            List<ReplyInbox.InboxMessage> got = awaitPeek(inbox, target);
+            assertEquals(1, got.size());
+            assertEquals("confirmed delivery", got.getFirst().content());
+            inbox.ack(target, "m1");
+        }
+    }
+
+    /** Poll peek until at least {@code n} replies for {@code target} are held, or ~10s elapse. */
+    @SuppressWarnings("BusyWait")
+    private static void awaitHeldAtLeast(AmqpReplyInbox inbox, String target, int n) throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        while (inbox.peek(target).size() < n && System.nanoTime() < deadline) {
+            Thread.sleep(50);
+        }
+    }
+
+    /**
+     * Repeatedly ack whatever is currently held (freeing prefetch slots for the next delivery) until
+     * both the local snapshot and the broker's own queue depth are empty, or ~10s elapse.
+     */
+    @SuppressWarnings("BusyWait")
+    private static void drainUntilEmpty(AmqpReplyInbox inbox, String target, Connection inspect) throws Exception {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        while (System.nanoTime() < deadline) {
+            for (ReplyInbox.InboxMessage msg : inbox.peek(target)) {
+                inbox.ack(target, msg.msgId());
+            }
+            try (Channel ch = inspect.createChannel()) {
+                if (ch.queueDeclarePassive(queueName(target)).getMessageCount() == 0 && inbox.peek(target).isEmpty()) {
+                    return;
+                }
+            }
+            Thread.sleep(50);
+        }
+        throw new AssertionError("did not drain " + target + " to empty within the deadline");
+    }
+
     /** Poll peek (broker delivery is async) until a reply for {@code target} appears or ~10s elapse. */
     @SuppressWarnings("BusyWait") // deliberate poll for async broker delivery, bounded by the deadline
     private static List<ReplyInbox.InboxMessage> awaitPeek(AmqpReplyInbox inbox, String target)

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

@@ -0,0 +1,264 @@
+package dev.ltms.bridged.msg;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConfirmCallback;
+import com.rabbitmq.client.Connection;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CB-528 follow-up: {@link AmqpReplyInbox#failPendingPublishesOnRecovery()} must not fail a publish
+ * that registers concurrently with (and was not yet visible when) the recovery sweep began — that
+ * would report a publish that actually succeeded as failed, and {@code MessageService.reply} retries
+ * with a fresh {@code msgId}, so the reply is delivered twice. A live broker reconnect cannot be
+ * forced reliably, so this drives {@link AmqpReplyInbox#failPendingPublishesOnRecovery} and a real
+ * {@link AmqpReplyInbox#publish} against each other directly, against fake AMQP channels built with
+ * {@link Proxy} (no mocking library is on the classpath).
+ *
+ * <p>Also covers Finding 2 (CB-528 follow-up): {@link AmqpReplyInbox#close()} must fail an in-flight
+ * publish promptly instead of leaving it to idle out the 10s confirm timeout.
+ */
+class AmqpReplyInboxRecoveryRaceTest {
+
+    /** Large enough that the (unfixed) unsynchronized sweep's iteration is a real, observable window
+     *  a concurrently-started publish can land in — not just a best case, single-entry sprint. */
+    private static final int STALE_PUBLISHES = 100_000;
+
+    @Test
+    @Timeout(30)
+    void recoverySweepDoesNotFailAPublishThatRegistersWhileItIsRunning() throws Exception {
+        AtomicLong seqCounter = new AtomicLong();
+        List<Long> seqOrder = new CopyOnWriteArrayList<>();
+        List<String> msgIdOrder = new CopyOnWriteArrayList<>();
+        AtomicReference<ConfirmCallback> ackCallback = new AtomicReference<>();
+        AtomicReference<ConfirmCallback> nackCallback = new AtomicReference<>();
+
+        Channel publishChannel = fakeChannel(seqCounter, seqOrder, msgIdOrder, ackCallback, nackCallback);
+        Channel consumeChannel = fakeChannel(seqCounter, seqOrder, msgIdOrder, ackCallback, nackCallback);
+        Connection connection = fakeConnection(consumeChannel, publishChannel);
+
+        AmqpReplyInbox inbox = new AmqpReplyInbox(connection, AmqpReplyInbox.DEFAULT_PREFETCH);
+
+        // STALE_PUBLISHES in-flight publishes that never get confirmed — they sit in pendingBySeq /
+        // pendingByMsgId exactly like publishes whose confirm never arrived before a connection drop.
+        // Virtual threads make this many concurrent blocking publish() calls cheap.
+        CountDownLatch staleStarted = new CountDownLatch(STALE_PUBLISHES);
+        for (int i = 0; i < STALE_PUBLISHES; i++) {
+            String msgId = "stale-" + i;
+            Thread.ofVirtual().start(() -> {
+                staleStarted.countDown();
+                try {
+                    inbox.publish("worker-stale", msgId, "x");
+                } catch (IllegalStateException expected) {
+                    // resolved (failed by the sweep) — that is exactly what this thread is here for
+                }
+            });
+        }
+        staleStarted.await();
+        // Let the registrations (the synchronized put into pendingBySeq/pendingByMsgId) actually land
+        // for all of them before the sweep starts, so the sweep begins with a large, real backlog.
+        Thread.sleep(300);
+
+        AtomicReference<Throwable> sweepError = new AtomicReference<>();
+        Thread sweepThread = new Thread(() -> {
+            try {
+                inbox.failPendingPublishesOnRecovery();
+            } catch (Throwable t) {
+                sweepError.set(t);
+            }
+        }, "recovery-sweep");
+        sweepThread.start();
+        // A short, deliberate head start: with STALE_PUBLISHES this large, the (unfixed) sweep's own
+        // iteration takes several milliseconds, so this guarantees the sweep has already begun —
+        // and, once guarded, is already holding publishChannelLock — before "fresh" attempts to
+        // register. Without this head start, "fresh" sometimes wins the race for the lock and
+        // registers before the sweep even starts, which is the accepted "already in flight when
+        // recovery fires" case (correctly failed either way) rather than the bug under test.
+        Thread.sleep(5);
+
+        // This is the exact interleaving CB-528's follow-up describes: "the still-running recovery
+        // sweep" racing a publish that registers while it is mid-flight.
+        AtomicReference<Throwable> publishError = new AtomicReference<>();
+        Thread freshThread = new Thread(() -> {
+            try {
+                inbox.publish("worker-fresh", "fresh", "hello");
+            } catch (Throwable t) {
+                publishError.set(t);
+            }
+        }, "fresh-publish");
+        freshThread.start();
+
+        sweepThread.join(20_000);
+        assertNull(sweepError.get(), "sweep threw: " + sweepError.get());
+
+        // Simulate the broker's real confirm for "fresh" now that the sweep is done, so a correct
+        // implementation's publish() returns normally instead of idling out CONFIRM_TIMEOUT_MS. Poll
+        // for the registration rather than checking once: freshThread may still be contending for
+        // publishChannelLock (behind the 20,000 stale threads' own lock acquisitions) even though the
+        // sweep itself has already finished.
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(9);
+        int idx = -1;
+        while (idx < 0 && System.nanoTime() < deadline) {
+            idx = msgIdOrder.indexOf("fresh");
+            if (idx < 0) {
+                Thread.sleep(20);
+            }
+        }
+        if (idx >= 0 && ackCallback.get() != null) {
+            ackCallback.get().handle(seqOrder.get(idx), false);
+        }
+        freshThread.join(15_000);
+
+        assertNull(publishError.get(),
+                "a publish that registered while the recovery sweep was running must not be failed by "
+                        + "it, but got: " + publishError.get());
+    }
+
+    @Test
+    @Timeout(15)
+    void closeFailsInFlightPublishPromptlyInsteadOfWaitingOutTheConfirmTimeout() throws Exception {
+        AtomicLong seqCounter = new AtomicLong();
+        List<Long> seqOrder = new CopyOnWriteArrayList<>();
+        List<String> msgIdOrder = new CopyOnWriteArrayList<>();
+        AtomicReference<ConfirmCallback> ackCallback = new AtomicReference<>();
+        AtomicReference<ConfirmCallback> nackCallback = new AtomicReference<>();
+
+        Channel publishChannel = fakeChannel(seqCounter, seqOrder, msgIdOrder, ackCallback, nackCallback);
+        Channel consumeChannel = fakeChannel(seqCounter, seqOrder, msgIdOrder, ackCallback, nackCallback);
+        Connection connection = fakeConnection(consumeChannel, publishChannel);
+
+        AmqpReplyInbox inbox = new AmqpReplyInbox(connection, AmqpReplyInbox.DEFAULT_PREFETCH);
+
+        CountDownLatch publishReturned = new CountDownLatch(1);
+        AtomicReference<Throwable> publishError = new AtomicReference<>();
+        AtomicLong elapsedMillis = new AtomicLong();
+        Thread publishThread = new Thread(() -> {
+            long start = System.nanoTime();
+            try {
+                inbox.publish("worker-close", "never-confirmed", "x");
+            } catch (Throwable t) {
+                publishError.set(t);
+            } finally {
+                elapsedMillis.set((System.nanoTime() - start) / 1_000_000);
+                publishReturned.countDown();
+            }
+        }, "publish-during-close");
+        publishThread.start();
+
+        Thread.sleep(200); // let publish() register before close() runs
+        inbox.close();
+
+        assertTrue(publishReturned.await(5, TimeUnit.SECONDS), "publish() did not return after close()");
+        assertNotNull(publishError.get(), "a publish in flight when close() runs must fail, not hang");
+        assertTrue(publishError.get().getMessage() != null
+                        && publishError.get().getMessage().toLowerCase().contains("closed"),
+                "expected a clear closed-inbox message, got: " + publishError.get());
+        assertTrue(elapsedMillis.get() < 5_000,
+                "close() should fail the in-flight publish promptly, not wait out the confirm timeout — took "
+                        + elapsedMillis.get() + "ms");
+    }
+
+    /** A {@link Proxy}-backed {@link Channel}: only the calls {@link AmqpReplyInbox} actually makes
+     *  are meaningfully implemented; everything else returns a harmless default. */
+    private static Channel fakeChannel(AtomicLong seqCounter, List<Long> seqOrder, List<String> msgIdOrder,
+                                        AtomicReference<ConfirmCallback> ackCallback,
+                                        AtomicReference<ConfirmCallback> nackCallback) {
+        InvocationHandler handler = (proxy, method, args) -> {
+            String name = method.getName();
+            if (name.equals("getNextPublishSeqNo")) {
+                long value = seqCounter.incrementAndGet();
+                seqOrder.add(value);
+                return value;
+            }
+            if (name.equals("basicPublish")) {
+                AMQP.BasicProperties props = (AMQP.BasicProperties) args[3];
+                msgIdOrder.add(props.getMessageId());
+                return null;
+            }
+            if (name.equals("addConfirmListener")) {
+                ackCallback.set((ConfirmCallback) args[0]);
+                nackCallback.set((ConfirmCallback) args[1]);
+                return null;
+            }
+            if (name.equals("equals")) {
+                return proxy == args[0];
+            }
+            if (name.equals("hashCode")) {
+                return System.identityHashCode(proxy);
+            }
+            if (name.equals("toString")) {
+                return "FakeChannel";
+            }
+            return defaultValue(method.getReturnType());
+        };
+        return (Channel) Proxy.newProxyInstance(AmqpReplyInboxRecoveryRaceTest.class.getClassLoader(),
+                new Class<?>[] {Channel.class}, handler);
+    }
+
+    /** A {@link Proxy}-backed {@link Connection} handing out {@code first} then {@code second} from
+     *  successive {@code createChannel()} calls, matching {@link AmqpReplyInbox}'s constructor. */
+    private static Connection fakeConnection(Channel first, Channel second) {
+        AtomicInteger calls = new AtomicInteger();
+        InvocationHandler handler = (proxy, method, args) -> {
+            String name = method.getName();
+            if (name.equals("createChannel") && (args == null || args.length == 0)) {
+                return calls.getAndIncrement() == 0 ? first : second;
+            }
+            if (name.equals("equals")) {
+                return proxy == args[0];
+            }
+            if (name.equals("hashCode")) {
+                return System.identityHashCode(proxy);
+            }
+            if (name.equals("toString")) {
+                return "FakeConnection";
+            }
+            return defaultValue(method.getReturnType());
+        };
+        return (Connection) Proxy.newProxyInstance(AmqpReplyInboxRecoveryRaceTest.class.getClassLoader(),
+                new Class<?>[] {Connection.class}, handler);
+    }
+
+    private static Object defaultValue(Class<?> type) {
+        if (!type.isPrimitive() || type == void.class) {
+            return null;
+        }
+        if (type == boolean.class) {
+            return Boolean.FALSE;
+        }
+        if (type == long.class) {
+            return 0L;
+        }
+        if (type == short.class) {
+            return (short) 0;
+        }
+        if (type == byte.class) {
+            return (byte) 0;
+        }
+        if (type == char.class) {
+            return (char) 0;
+        }
+        if (type == double.class) {
+            return 0.0d;
+        }
+        if (type == float.class) {
+            return 0.0f;
+        }
+        return 0;
+    }
+}

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

@@ -81,7 +81,7 @@ class ReplyPushLoopTest {
     @Test
     void decideWithNothingPendingIsStop() {
         agents = agentWithStatus("idle");
-        assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -90,7 +90,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop(2, 100);
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2, 0));
     }
 
     @Test
@@ -99,7 +99,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -108,7 +108,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0),
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0),
                 "BLOCKED is injectable");
     }
 
@@ -118,7 +118,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0),
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0),
                 "DONE is injectable");
     }
 
@@ -128,7 +128,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -137,7 +137,7 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -146,9 +146,9 @@ class ReplyPushLoopTest {
         inbox.publish(WORKER, "m1", "hello");
         var loop = loop();
         loop.onReplyQueued(WORKER);
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0));
         inbox.ack(WORKER, "m1");
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 0));
     }
 
     // --- onReplyQueued integration -------------------------------------------------------------
@@ -240,7 +240,7 @@ class ReplyPushLoopTest {
     @Test
     void decideTicketsWithNothingPendingIsStop() {
         agents = agentWithStatus("idle");
-        assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -248,7 +248,7 @@ class ReplyPushLoopTest {
         agents = agentWithStatus("idle");
         var loop = loop(2, 100_000);
         loop.onTicketTerminal("task-1", WORKER, false);
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 2));
     }
 
     @Test
@@ -256,7 +256,7 @@ class ReplyPushLoopTest {
         agents = agentWithStatus("idle");
         var loop = loop(5, 100_000);
         loop.onTicketTerminal("task-1", WORKER, false);
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -264,7 +264,7 @@ class ReplyPushLoopTest {
         agents = agentWithStatus("working");
         var loop = loop(5, 100_000);
         loop.onTicketTerminal("task-1", WORKER, false);
-        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0, 0));
     }
 
     @Test
@@ -272,7 +272,7 @@ class ReplyPushLoopTest {
         agents = agentWithStatus("idle");
         var loop = new ReplyPushLoop(new PrimaryRegistry(null), agents, inbox, scheduler, 5, 100_000);
         loop.onTicketTerminal("task-1", WORKER, false); // no lead known -> never registered as pending
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 0));
     }
 
     // --- CB-588: onTicketTerminal integration ---------------------------------------------------
@@ -420,6 +420,49 @@ class ReplyPushLoopTest {
                 + "restart the loop — that would defeat the reminder cap");
     }
 
+    // --- mirror of the two stopOrRestart tests above, for the reply arm ------------------------
+    //
+    // Both tests above only ever passed Set.of() for repliesBefore, so racedIn's reply branch
+    // (`pendingReplyTargetsFor(lead).stream().anyMatch(t -> !repliesBefore.contains(t))`) was
+    // never exercised by anything other than an always-empty snapshot. The reviewer flagged this:
+    // racedIn is symmetric in the code, and only half of it was pinned by a test.
+
+    @Test
+    void aReplyStillPendingWhenTheLoopStopsIsNotStranded() {
+        // Mirrors aTicketStillPendingWhenTheLoopStopsIsNotStranded: a reply target that raced in
+        // during the decision-to-release window (absent from the "before" snapshot) must reclaim
+        // the schedule slot rather than being stranded with no schedule left to nudge about it.
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        inbox.publish(WORKER, "m1", "hello");
+        ReplyPushLoop loop = loop(1, 100_000); // long backoff — no natural tick fires during this test
+
+        loop.onReplyQueued(WORKER); // pendingReplies={term_worker}; activeLeads={PRIMARY}
+
+        loop.stopOrRestart(PRIMARY, Set.of(), Set.of());
+
+        assertTrue(loop.isActive(), "a reply that raced the loop's stop must reclaim the schedule "
+                + "slot, not be stranded with no schedule left to ever nudge about it");
+    }
+
+    @Test
+    void aStaleUncollectedReplyAtCapDoesNotRestartTheLoop() {
+        // Mirrors aStaleUncollectedTicketAtCapDoesNotRestartTheLoop: a reply target already
+        // accounted for at decide time (present in repliesBefore) must not restart the loop —
+        // that is the reminder cap doing its job, not a race.
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        inbox.publish(WORKER, "m1", "hello");
+        ReplyPushLoop loop = loop(1, 100_000);
+
+        loop.onReplyQueued(WORKER); // pendingReplies={term_worker}; activeLeads={PRIMARY}
+
+        loop.stopOrRestart(PRIMARY, Set.of(WORKER), Set.of());
+
+        assertFalse(loop.isActive(), "a stale reply target already accounted for at decide time "
+                + "must not restart the loop — that would defeat the reminder cap");
+    }
+
     @Test
     void ticketNudgeFormatIsCorrect() {
         String single = ReplyPushLoop.TICKET_NUDGE_FORMAT.formatted("task-1", "", "task-1");
@@ -485,6 +528,57 @@ class ReplyPushLoopTest {
         assertTrue(nudge.contains("task-1"), "the ticket must not be dropped: " + nudge);
     }
 
+    // --- CB-590 follow-up: per-source reminder budgets — the regression this round exists for ---
+
+    @Test
+    void oneExhaustedSourceDoesNotBlockANudgeForTheOtherSource() {
+        // The live trace this ticket was filed from: an undrained reply target got nudged up to
+        // its cap (5 reminders), then a ticket for the SAME lead went terminal shortly before the
+        // next scheduled tick — so it coalesced onto the still-active schedule (arriving BEFORE
+        // that tick's "before" snapshot, not during the decision-to-release race stopOrRestart
+        // guards). With CB-590's single shared reminder counter, that tick's decide() saw
+        // reminderCount already at the cap and returned STOP regardless of the ticket, and because
+        // the ticket was already present in that tick's "before" snapshot, stopOrRestart's
+        // racedIn check (proven correct on its own above) did not save it either — it is not a
+        // race, it looks like ordinary stale backlog. The ticket was then stranded: pending
+        // forever with no live schedule, never named in any nudge.
+        //
+        // Fixed by giving each source its own counter. Here the reply source is AT its cap (2/2)
+        // and the ticket source has NEVER been nudged (0/2) — decide() must still return INJECT,
+        // because the ticket is still eligible on its own budget.
+        agents = agentWithStatus("idle");
+        inbox.publish(WORKER, "m1", "hello");
+        var loop = loop(2, 100_000); // huge backoff — this test drives decide()/isActive() directly
+        loop.onReplyQueued(WORKER);
+        loop.onTicketTerminal("task-1", WORKER, false);
+
+        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 2, 0),
+                "the reply source is exhausted (2/2), but the ticket source has never been "
+                        + "nudged (0/2) — the lead must still be injected so the ticket is not "
+                        + "lost, exactly the CB-590 follow-up regression");
+
+        // isActive() (criterion #5): a real tick that takes the INJECT branch above never calls
+        // stopOrRestart, so the schedule started by onTicketTerminal above stays live — the
+        // ticket is not left stranded with isActive()==false while it is still pending.
+        assertTrue(loop.isActive(), "the schedule must stay active while the ticket source still "
+                + "has budget left, even though the reply source sharing it is exhausted");
+    }
+
+    @Test
+    void bothSourcesExhaustedIsStillStop() {
+        // The flip side: per-source budgets must not turn into unbounded nudging. When BOTH
+        // sources are at their cap, decide() must still STOP — a per-source budget is still a
+        // budget.
+        agents = agentWithStatus("idle");
+        inbox.publish(WORKER, "m1", "hello");
+        var loop = loop(2, 100_000);
+        loop.onReplyQueued(WORKER);
+        loop.onTicketTerminal("task-1", WORKER, false);
+
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2, 2),
+                "both the reply and the ticket source are at their own cap — must still stop");
+    }
+
     // --- metrics (CB-512) ----------------------------------------------------------------------
 
     @Test
@@ -513,7 +607,7 @@ class ReplyPushLoopTest {
         var loop = loop(2, 100, metrics);
         loop.onReplyQueued(WORKER);
 
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2, 0));
 
         assertEquals(1, metrics.count(BridgedMetrics.PUSH_NUDGES, "outcome", "exhausted"),
                 "hitting the reminder cap must count as exhausted");
@@ -541,7 +635,7 @@ class ReplyPushLoopTest {
         var loop = loop(2, 100_000, metrics);
         loop.onTicketTerminal("task-1", WORKER, false);
 
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 2));
 
         assertEquals(1, metrics.count(BridgedMetrics.PUSH_NUDGES, "outcome", "exhausted"),
                 "hitting the ticket reminder cap must count as exhausted");