CB-527/CB-528: bound AMQP prefetch and confirm publishes before claiming durable

CB-527: basicQos(prefetch) on the consume channel before basicConsume, configurable
via broker.prefetch (default 32), so an undrained inbox backlog stays on the broker
instead of growing the JVM heap without limit.

CB-528: publish moves to its own confirm-mode channel with mandatory=true and a
return listener, so an unroutable or unconfirmed reply now throws instead of
vanishing silently. The confirm callback checks the per-message returned flag
(set by the return listener, which the broker always fires before the matching
confirm) so an acked-but-returned publish is still reported as a failure. The ack
path stays on its own channel/lock and never waits on a publish confirm.

CLAUDE.md

@@ -81,9 +81,9 @@ below are the procedure — run them in order, every task, not only the big ones
 5. **Collect** — `bridge_poll{ticket}` → `bridge_ack{ticket, msgId}`. Answer a worker's `bridge_ask`
    with `bridge_send{turnId, content}` — **not** `sessionId`. A worker gone quiet is diagnosed with
    `bridge_status`, never by reading its terminal.
-6. **Verify yourself.** Re-run the build and the checks. A worker mounts only the bridge MCP and
-   cannot run your other tooling, and a piped command (`… | tail`) hides failures behind a zero
-   exit — never promote a worker's "clean" to a fact.
+6. **Verify yourself.** Re-run the build and the checks. A worker cannot run your IDE tooling, any
+   forge tools it appears to have hold a blocked credential and fail, and a piped command
+   (`… | tail`) hides failures behind a zero exit — never promote a worker's "clean" to a fact.
 7. **Review — fan out.** Spawn reviewers against the diff, one per dimension or per file, with
    `wait:false`. Never the implementer of the scope it reviews, and brief them from the diff — not
    from the implementer's rationale, which carries its own blind spot. Dispatch each PR's reviewers
@@ -155,9 +155,13 @@ you.
    without replying, the bridge scrapes your pane, and it can return only the last 4000 characters.
    A clipped scrape is marked as partial, but the missing text is gone — your report reaches the
    lead with its end cut off.
-5. **Report honestly.** State only what you actually ran and its real output, including failures.
-   You mount **only** the bridge MCP — the primary's other servers (IDE, forge, docs) are not yours,
-   so never claim the result of a check you had no way to run.
+5. **Report honestly.** State only what you actually ran and its real output, including failures,
+   and never claim the result of a check you had no way to run. **Measure your own tools; do not
+   assume them.** What you mount depends on your backend: an opencode member gets the bridge and
+   nothing else, while a Claude Code member also inherits the operator's user-scope MCP servers,
+   which the bridge never chose for you. Two rules follow. The primary's IDE tooling is still not
+   yours, whatever you see. And **a mounted tool is not a working tool** — the forge server you may
+   find there holds a deliberately blocked credential and fails every call, by design.
 6. **Never merge.** Stage files explicitly — never `git add -A` — and leave alone anything the
    project marks as not-yours-to-commit.
 

bridged/bridged.example.yaml

@@ -434,10 +434,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/member/HerdrPeerLauncher.java

@@ -781,10 +781,43 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      * PATH} seeding already depends on the overlay reliably replacing an inherited value (see its
      * javadoc), and that is only demonstrated for a non-blank value, so this reuses the same,
      * proven-reliable shape rather than the unverified one.
+     *
+     * <p><b>MEASURED ON A LIVE PANE, 2026-08-15: this sentinel alone does NOT hold.</b> The overlay
+     * itself works — {@code GITEA_TOKEN} is injected here, is exported by no shell file, and does
+     * reach the pane. The sentinel loses one step later. A herdr pane runs a <em>login</em> shell,
+     * {@code ~/.zprofile} sources {@code ${SHARED_ENV}/tools/secrets.sh}, and that file does a plain
+     * unconditional {@code export GITEA_ACCESS_TOKEN=...}. A login shell overwrites a value already
+     * in the environment, so the real admin token is put back over this sentinel before the member
+     * process ever starts. That defeat applies to <em>every</em> name {@code secrets.sh} exports,
+     * and no launcher-side overlay can win against it.
+     *
+     * <p>So this constant is not the control on its own — {@link #MEMBER_MARKER} is the other half.
+     * Keeping the sentinel is still worth it: it is correct for any peer kind whose pane does not
+     * start a login shell, and it makes the intent explicit at the one place every adapter passes.
      */
     private static final String BLOCKED_GITEA_ACCESS_TOKEN =
             "blocked-by-bridged-cb592-see-gitea-issue-77";
 
+    /**
+     * CB-592: marks a pane as a bridged member so a shell startup file can decline to export
+     * operator-only credentials into it (gitea issue #77).
+     *
+     * <p>This name is deliberately one that {@code secrets.sh} never exports, which is exactly why
+     * it survives the login shell that wipes {@link #BLOCKED_GITEA_ACCESS_TOKEN}. The mechanism is
+     * measured, not assumed: {@code GITEA_TOKEN} is injected the same way, is absent from a login
+     * shell of its own, and was observed set inside a live member pane.
+     *
+     * <p>It is a no-op until the operator guards the export, which is a one-line change in a file
+     * this repo does not own and must not edit unasked:
+     *
+     * <pre>{@code
+     * [ -n "${BRIDGED_MEMBER:-}" ] || export GITEA_ACCESS_TOKEN=...
+     * }</pre>
+     *
+     * <p>Setting the marker now costs nothing and means that edit is the whole remaining fix.
+     */
+    static final String MEMBER_MARKER = "BRIDGED_MEMBER";
+
     /**
      * Seed a worker's environment (CB-511): the daemon's own {@code PATH}, then the profile's
      * {@code env:} entries, then the CB-592 admin-token shadow.
@@ -802,10 +835,11 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      * dev.ltms.bridged.guard.SubscriptionGuard}, which is checked against the profile's
      * {@code baseUrl} and nothing else.
      *
-     * <p>The CB-592 shadow is put in <em>last</em>, after the profile's own {@code env:}, so no
-     * profile — present or future — can restore the admin token by naming it in config. This is
-     * the one place the shadow is applied: every {@code buildLaunch} in every adapter calls this
-     * first, so a new profile, and a peer kind not yet written, gets it for free.
+     * <p>The CB-592 shadow and marker are put in <em>last</em>, after the profile's own
+     * {@code env:}, so no profile — present or future — can restore the admin token, or hide that
+     * the pane is a member, by naming either in config. This is the one place both are applied:
+     * every {@code buildLaunch} in every adapter calls this first, so a new profile, and a peer
+     * kind not yet written, gets them for free.
      */
     protected Map<String, String> baseEnv(BridgedConfig.Profile cfg) {
         Map<String, String> workerEnv = new LinkedHashMap<>();
@@ -817,6 +851,7 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
             workerEnv.putAll(cfg.env());
         }
         workerEnv.put("GITEA_ACCESS_TOKEN", BLOCKED_GITEA_ACCESS_TOKEN);
+        workerEnv.put(MEMBER_MARKER, "1");
         return workerEnv;
     }
 

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,81 @@ 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. */
+    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 +217,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 +230,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,6 +327,65 @@ 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. */
+    private void failPendingPublishesOnRecovery() {
+        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"));
+        }
+    }
+
     private static String queueName(String target) {
         return QUEUE_PREFIX + target + QUEUE_SUFFIX;
     }
@@ -233,6 +397,11 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
         } 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/test/java/dev/ltms/bridged/member/ClaudeCodeLauncherTest.java

@@ -685,6 +685,12 @@ class ClaudeCodeLauncherTest {
      * explicit (non-blank) GITEA_ACCESS_TOKEN to herdr on every spawn, whatever the profile is, so
      * a future baseEnv refactor cannot silently drop it and reopen the leak. Asserted against what
      * tab.create's params actually carry, not an internal map built in the test (gitea #77).
+     *
+     * <p>Scope, measured on a live pane 2026-08-15: this pins what the launcher SENDS, and that is
+     * all it can pin. It does not prove the value survives, and it does not: the pane runs a login
+     * shell, ~/.zprofile sources secrets.sh, and its unconditional `export GITEA_ACCESS_TOKEN=...`
+     * puts the real token back over this sentinel. Closing that needs the operator to guard the
+     * export on BRIDGED_MEMBER — see everySpawnMarksThePaneAsAMember below.
      */
     @Test
     void everySpawnShadowsTheAdminGiteaAccessToken() {
@@ -716,6 +722,39 @@ class ClaudeCodeLauncherTest {
                 "a profile's own env: must not be able to smuggle the admin token back in");
     }
 
+    /**
+     * The half of CB-592 that can actually survive the pane's login shell. BRIDGED_MEMBER is a name
+     * secrets.sh never exports, so nothing overwrites it — measured: GITEA_TOKEN is injected the
+     * same way, is absent from a login shell of its own, and was observed set inside a live member
+     * pane. It lets the operator guard the admin export with
+     * `[ -n "${BRIDGED_MEMBER:-}" ] || export GITEA_ACCESS_TOKEN=...`, which is the whole fix.
+     * Pinned here so a refactor cannot drop the marker and quietly un-guard every member (#77).
+     */
+    @Test
+    void everySpawnMarksThePaneAsAMember() {
+        FakeHerdr herdr = new FakeHerdr();
+        service(herdr, List.of("claude"), null).spawn();
+
+        assertEquals("1", startEnv(herdr).get("BRIDGED_MEMBER"),
+                "every member pane must be marked, or a shell file cannot tell it apart from the operator's");
+    }
+
+    /** A profile must not be able to hide that its pane is a member, for the same reason as above. */
+    @Test
+    void aProfileEnvEntryCannotClearTheMemberMarker() {
+        FakeHerdr herdr = new FakeHerdr();
+        BridgedConfig.Profile cfg = new BridgedConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN",
+                List.of("claude"), "tab", "bridged-workers", "w #{n}", null, null, null, null, null,
+                null, Map.of("BRIDGED_MEMBER", ""), null, null);
+        new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
+                _ -> null).spawn();
+
+        assertEquals("1", startEnv(herdr).get("BRIDGED_MEMBER"),
+                "a profile's own env: must not be able to unmark its pane");
+    }
+
     // ── CB-533: the model is pinned on the command line, not only in the environment ────────────
 
     /** A launcher for a profile identical but for its {@code model:} — the only variable here. */

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)

docs/CB-591-Gateway-Migration.md

@@ -1,6 +1,11 @@
 # CB-591 — move the fleet onto the LLM and MCP gateway
 
-**Status:** plan, not started · **Upstream:** [systems/vms wiki → LLM and MCP Gateway](https://git.ltms.dev/systems/vms/wiki/LLM-and-MCP-Gateway)
+**Status: DONE — the fleet is on the gateway as of 2026-08-15.** `local` runs on `/anthropic` and
+`gx` on `/v1`, both at `weight: 100`; `local-direct` stays at `weight: 0` as the escape hatch. Getting
+here took a revert and two upstream fixes — see §7.1, which is the useful part of this document. One
+risk is **accepted rather than solved**: a stream cut by any mid-response timer arrives as HTTP 200
+with no terminator, and our third-party members cannot detect it (§7.2).
+· **Upstream:** [systems/vms wiki → LLM and MCP Gateway](https://git.ltms.dev/systems/vms/wiki/LLM-and-MCP-Gateway)
 · **Upstream issue:** [systems/vms#31](https://git.ltms.dev/systems/vms/issues/31)
 
 The gateway went live on 2026-08-15 and replaced Bifrost. This plan says what that means for a
@@ -154,12 +159,21 @@ speaks the Anthropic protocol, so `/anthropic` is both correct and the only safe
 This is the exact failure shape this repo keeps hitting: it compiles, it answers, it looks healthy,
 and a capability is quietly off. Treat it as a `silent-default` risk, not a config preference.
 
-**Open question for 3b, stated as open.** The opencode profile must use the OpenAI surface, because
-that is the only thing an OpenAI-compatible provider can speak. vLLM serves that API natively, so I
-*expect* it to be a passthrough as well — but the wiki documents the translation trap only for the
-Anthropic path, and I have not checked how `reasoning_content` behaves through `/v1`. Verify it on
-first spawn (§7) rather than assuming. If reasoning is dropped there, that is a limitation of the
-opencode profile, not a reason to abandon it — opencode members do dev work, not deep reasoning.
+**Open question for 3b — ANSWERED, 2026-08-15.** The worry was that the OpenAI surface might drop
+reasoning the way the wiki documents for a mis-declared Anthropic backend. It does not. Checked at
+the API before any profile was switched:
+
+| surface | request | result |
+|---|---|---|
+| `/anthropic/v1/messages` | `deepseek-v4-flash`, 64 tokens | 200, response carries a real `"type":"thinking"` block |
+| `/v1/chat/completions` | same | 200, message carries a populated `reasoning_content` (and a `reasoning` field) |
+| `/v1/models` | — | 200, exactly `["deepseek-v4-flash"]` — the exact-name trap is clear |
+| `/v1/models`, **no token** | — | **401** — Caddy is gating, as designed |
+
+So reasoning survives on **both** surfaces, and the `/anthropic` choice for `local` is about protocol
+correctness rather than a repair for a known loss. The last row matters on its own: the wiki warns
+the gateway's own `SecurityPolicy` fails open, so it is worth knowing the proxy in front really does
+refuse an unauthenticated request here.
 
 ---
 
@@ -290,6 +304,160 @@ Merging config is not proving it. The checks, in order:
 
 ---
 
+## 7.1 What the live run actually found — 2026-08-15
+
+U1–U2c were done, the daemon restarted onto them, and both new profiles were spawned for real. The
+migration was then **reverted**. This section is the result, so none of it has to be re-derived.
+
+### The blocker
+
+`llm.ltms.dev` answers **HTTP 413 Request Entity Too Large** above **32 KiB (32768 bytes)**, on both
+surfaces:
+
+```
+/v1        32695 bytes -> 200        /anthropic  32095 bytes -> 200
+/v1        32795 bytes -> 413        /anthropic  32855 bytes -> 413
+```
+
+32 KiB is far below one real agent turn.
+
+**Root cause — confirmed by the systems/vms side, 2026-08-15.** My guess that it was a Caddy
+`request_body max_size` was **wrong**. It is Envoy, inside `aigw` on `llm.vm`. Envoy Gateway defaults
+a listener's `per_connection_buffer_limit_bytes` to **32768**, and the AI Gateway buffers the *whole*
+request body before it can route on the model name — so that default is not a network tuning knob
+here, it is a hard ceiling on prompt size. Read out of the live Envoy `config_dump`:
+
+```
+listener default/llm/http    per_connection_buffer_limit_bytes: 32768
+```
+
+Nobody chose 32 KiB; it was inherited from the default. Both TLS edges are innocent: the same
+boundary reproduces on the LAN path and the internet path, and both 413s carry an `x-llm-consumer`
+header their auth proxy sets only *after* authenticating — so the body cleared both edges and the
+auth. Directly on `llm.vm`, `aigw` 413s at 39 KB while the vLLM backend accepts the same 39 KB and
+answers 200.
+
+**Do not plan around 32 KiB.** The intended ceiling is far higher. Their fix — a `ClientTrafficPolicy`
+setting `bufferLimit: 8Mi` — is written but **not deployed** as of this note, pending their operator's
+approval. I have not re-tested and will not until they confirm, so as not to measure a half-changed
+system. Fixed in **systems/vms**, not here.
+
+### The part worth remembering
+
+Two members were spawned at the same moment with the same message:
+
+| | `local` (claude-code, `/anthropic`) | `gx` (opencode, `/v1`) |
+|---|---|---|
+| READY → BUSY | 19:07:26 | 19:07:45 |
+| BUSY → DONE | **19:08:51 (66s)** | **never — 10+ min, ticket FAILED** |
+
+**`local` passed.** It passed only because the probe was three trivial questions in a fresh session,
+so the request fit under 32 KiB. The profile looked healthy and was a landmine set to fire on the
+first turn that reads a file.
+
+So §7's checklist was not wrong, it was **too easy**. Any future run of it must use a task that reads
+a real file. A liveness probe proves the token and the URL; it does not prove the path.
+
+`gx` did not fail loudly either. Reproduced outside the bridge by running `opencode` by hand with the
+launcher's own generated config:
+
+```
+Error: Request Entity Too Large
+...compacts context, retries...
+Error: Request Entity Too Large
+```
+
+opencode **catches the 413, compacts, and retries — indefinitely**. A member that fails loudly costs
+one turn; this one costs the whole task and is indistinguishable from a slow worker.
+
+> **Diagnosing a stuck opencode member.** Do not read its pane. The launcher writes its config to a
+> temp dir and passes it as `OPENCODE_CONFIG` — find it with
+> `ls -dt /var/folders/*/*/T/bridged-opencode-* | head -1`, check the provider block and the key's
+> length and prefix (never its value), then reproduce with `opencode run --auto -m <provider>/<model>`
+> using the same `OPENCODE_CONFIG`. That is what turned "it hangs" into a one-line error.
+
+### What checked out, and needs no re-testing
+
+- Token accepted on both surfaces. **Unauthenticated → 401**, so the Caddy proxy really does gate —
+  the wiki's "SecurityPolicy fails open" warning is about the gateway itself, not the edge.
+- `/v1/models` returns exactly `["deepseek-v4-flash"]`, so trap 3 is clear.
+- **Reasoning survives both surfaces** — see §3b above.
+- The launcher's generated opencode provider block is correct, carrying a real 48-character `llmk-`
+  key rather than the `bridged-local-noauth` placeholder.
+- `SubscriptionGuard` accepted `llm.ltms.dev` after the allowlist edit and the restart: `local`
+  spawned without throwing, which is the check that catches a missed restart.
+
+### Resolution — both ceilings fixed, migration completed
+
+systems/vms fixed both, and each was re-checked from this side rather than taken on trust:
+
+| ceiling | was | now | our own check |
+|---|---|---|---|
+| listener buffer | 32 KiB | 32 Mi | 1.2 MB body → **200** (was 413) |
+| LLM route timeout | 60s | 86400s | the request that truncated: **101s, `message_stop` present, 4000/4000** |
+
+The timeout moved in two steps on 2026-08-15: 60s → 1800s, then 1800s → **86400s (24 hours)** after
+the truncation risk below was discussed. They tried `request: 0s` first, which removes the
+total-duration timer completely. It works, but on an `AIGatewayRoute` the **idle timeout is derived
+from the request timeout**, so `0s` also removed any bound on a stalled connection. 86400s keeps a
+reaper for dead connections while putting the truncation timer out of practical reach.
+
+Neither was deliberate. The 32 KiB was Envoy Gateway's default `per_connection_buffer_limit_bytes`;
+the 60s was Envoy AI Gateway's own documented default. The 60s bounded **generation** as well as
+prompt size — a tiny prompt with a long answer returned 504 at 60.05s.
+
+Two configuration facts worth keeping, from their bisection:
+
+- **`ClientTrafficPolicy` is honoured in standalone `aigw run`; `BackendTrafficPolicy` is NOT.** A
+  `BackendTrafficPolicy` setting `requestTimeout` is accepted, logs nothing, and leaves the routes
+  unchanged (upstream `envoyproxy/gateway#9513`). What works is `timeouts: {request: …}` on each
+  `AIGatewayRoute` rule. Nothing from the outside distinguishes the two — the same silent-default
+  shape as their `SecurityPolicy` caveat.
+- In that stack, "the config was accepted" proves nothing. Read the live `config_dump`.
+
+## 7.2 The risk we accepted, and why we could not remove it
+
+Raising the timeout made the failure **rare, not impossible**, and the residual failure is silent.
+
+On a mid-response timeout over chunked HTTP/1.1, Envoy ends the chunked encoding *cleanly* instead of
+resetting the connection, so the client receives what looks like a complete transfer
+(`envoyproxy/envoy#17186` — acknowledged as a bug in 2021, closed by a stale bot, never fixed). The
+December 2025 fix `envoyproxy/envoy#42269` changes locally-originated resets from `NO_ERROR` to
+`INTERNAL_ERROR`, but it is **HTTP/2 only** and SSE clients here speak HTTP/1.1.
+
+Measured on our side while the timeout was still 60s:
+
+```
+HTTP 200   61.07s   141992 bytes
+  message_stop 0    message_delta 0    error events 0
+  emitted 2473 of 4000, ending on a WELL-FORMED SSE frame
+```
+
+A syntactically valid stream that simply stops. Any timer firing mid-stream — route timeout, idle
+timeout, `max_stream_duration` — fails this same way.
+
+**The recommended defence does not transfer to us.** The right fix is to treat a stream with no
+`message_stop` / `[DONE]` / `finish_reason` as failed. We cannot: our members are Claude Code and
+opencode, third-party clients whose SSE parsing we do not own, and there is no seam to insert the
+check. Whether either detects a missing terminator is unverified — and opencode's handling of the 413
+(swallow, compact, retry forever, never surface an error) does not suggest it is strict.
+
+So the honest statement of our position:
+
+> Gateway traffic is acceptable at 86400s because a single request would have to run for 24 hours to
+> trip the bug — **not** because we could detect it if it did.
+
+At 86400s our **own** limit binds first, which is the ordering we want. `MessageService.ASYNC_TIMEOUT_MS`
+caps a turn at 30 minutes, so a runaway request ends as a clean `FAILED` ticket that we raised, rather
+than as a silently truncated `200` that we cannot see. While the gateway sat at 1800s the two numbers
+were equal and did not nest, so a gateway-side stall could have been misread as a bug in our own ticket
+handling. That ambiguity is now gone.
+
+**If a member ever returns a confident but truncated answer, suspect this before anything in our own
+code.** That is the whole reason this section exists.
+
+---
+
 ## 8. Related
 
 - [CB-589 / #74](https://git.ltms.dev/lms/claude-bridge/issues/74) — cost-first placement and a