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.
2026-08-16 16:55:40 +02:00
14 changed files with 283 additions and 481 deletions
@@ -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
@@ -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)");
@@ -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;
+        }
    }

    /**
@@ -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) {
@@ -234,14 +234,7 @@ public final class BridgedApp {
        List<Map<String, Object>> out = sessions.roster().stream()
                .map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
                .toList();
-        Map<String, Object> body = new LinkedHashMap<>();
-        body.put("workers", out);
-        // CB-586: operator visibility for the refs/wip snapshot store without shelling into the
-        // repo — how many snapshot refs exist and roughly what they cost. Present only once a
-        // worktree session has established the repo, so a never-snapshotted fleet reports nothing.
-        sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
-                Map.of("count", st.count(), "costBytes", st.costBytes())));
-        ctx.status(200).json(body);
+        ctx.status(200).json(Map.of("workers", out));
    }

    /** The configured worker profiles and which one a no-argument spawn uses. */
@@ -12,12 +12,9 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.security.SecureRandom;
-import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -302,129 +299,6 @@ public final class GitWorktrees implements Worktrees {
        return index;
    }

-    /**
-     * One {@code refs/wip/<branch>} snapshot ref as read by {@link #listWipRefs}: its full ref name,
-     * the snapshot commit's sha, and that commit's committer time in unix millis (the age of the
-     * snapshot — a snapshot is written once and never rewritten, so the commit date is the ref's).
-     */
-    private record WipRef(String refName, String sha, long committerMillis) {
-        String branch() {
-            return refName.substring("refs/wip/".length());
-        }
-    }
-
-    @Override
-    public WipRefStats wipRefs(String repoRoot) {
-        List<WipRef> refs = listWipRefs(repoRoot);
-        long costBytes = 0;
-        for (WipRef ref : refs) {
-            costBytes += treeSize(repoRoot, ref.sha());
-        }
-        return new WipRefStats(refs.size(), costBytes);
-    }
-
-    @Override
-    public int pruneWipRefs(String repoRoot, long minAgeMillis) {
-        // The rule is documented on Worktrees#pruneWipRefs: delete only a snapshot whose tree
-        // content is already reachable from main AND that is older than minAgeMillis. Reachability
-        // is the floor that keeps a worker's last copy; the age floor keeps a just-written snapshot
-        // from being swept while a lead may still be looking at it.
-        List<WipRef> refs = listWipRefs(repoRoot);
-        if (refs.isEmpty()) {
-            return 0;
-        }
-        long nowMillis = System.currentTimeMillis();
-        // Resolve what main carries once per sweep, not once per ref.
-        Set<String> mainObjects = reachableObjectsFromMain(repoRoot);
-        int deleted = 0;
-        for (WipRef ref : refs) {
-            long ageMillis = nowMillis - ref.committerMillis();
-            if (ageMillis <= minAgeMillis) {
-                continue; // too recent — never swept, even if it looks recoverable (CB-586)
-            }
-            String tree = exec("git", "-C", repoRoot, "rev-parse", ref.sha() + "^{tree}").trim();
-            if (!mainObjects.contains(tree)) {
-                // Last copy of the snapshot's content — the worker's work exists nowhere else.
-                // Never delete automatically (CB-586 criterion 2).
-                continue;
-            }
-            exec("git", "-C", repoRoot, "update-ref", "-d", ref.refName());
-            deleted++;
-            log.info("pruned snapshot ref refs/wip/{} commit={} (age {}h): its tree is already "
-                     + "reachable from main, so the work is preserved; recover from reflog via "
-                     + "git update-ref refs/wip/{} {}",
-                    ref.branch(), ref.sha(), TimeUnit.MILLISECONDS.toHours(ageMillis),
-                    ref.branch(), ref.sha());
-        }
-        return deleted;
-    }
-
-    /**
-     * Every {@code refs/wip/*} ref (see {@link WipRef}). The committer date is read as a unix
-     * count of seconds and converted to millis. {@code %00} (NUL) separates the fields because a
-     * branch name may contain spaces.
-     */
-    private List<WipRef> listWipRefs(String repoRoot) {
-        String out = exec("git", "-C", repoRoot, "for-each-ref",
-                "--format=%(refname)%00%(objectname)%00%(committerdate:unix)", "refs/wip/");
-        List<WipRef> refs = new ArrayList<>();
-        for (String line : out.split("\\R")) {
-            if (line.isBlank()) {
-                continue;
-            }
-            String[] parts = line.split("\u0000", -1);
-            if (parts.length == 3 && !parts[1].isBlank()) {
-                refs.add(new WipRef(parts[0], parts[1], Long.parseLong(parts[2]) * 1000L));
-            }
-        }
-        return refs;
-    }
-
-    /**
-     * The set of object shas reachable from {@code main}, or an empty set when {@code main} cannot
-     * be resolved. An empty set is the safe direction: the retention sweep then concludes nothing
-     * is recoverable, so it deletes nothing — a repo with no {@code main} must never cause a
-     * worker's last copy of a snapshot to be dropped on a reachability misreading.
-     */
-    private Set<String> reachableObjectsFromMain(String repoRoot) {
-        if (exitCode("git", "-C", repoRoot, "rev-parse", "--verify", "main") != 0) {
-            log.debug("refs/wip retention: no 'main' ref in {} — treating nothing as reachable", repoRoot);
-            return Set.of();
-        }
-        String out = exec("git", "-C", repoRoot, "rev-list", "--objects", "main");
-        Set<String> objects = new HashSet<>();
-        for (String line : out.split("\\R")) {
-            if (line.isBlank()) {
-                continue;
-            }
-            int sp = line.indexOf(' ');
-            objects.add(sp < 0 ? line : line.substring(0, sp));
-        }
-        return objects;
-    }
-
-    /** Approximate cost of a snapshot: the sum of every blob's size in its committed tree. */
-    private long treeSize(String repoRoot, String sha) {
-        String out = exec("git", "-C", repoRoot, "ls-tree", "-r", "-l", sha);
-        long total = 0;
-        for (String line : out.split("\\R")) {
-            if (line.isBlank()) {
-                continue;
-            }
-            // ls-tree -l row: "<mode> <type> <object> <size>\t<path>"; the size is only numeric for
-            // blobs (trees read "-"), so gate on the type token and take the 4th whitespace field.
-            String[] parts = line.split("\\s+");
-            if (parts.length >= 4 && "blob".equals(parts[1])) {
-                try {
-                    total += Long.parseLong(parts[3]);
-                } catch (NumberFormatException ignored) {
-                    // a '-' size (or any anomaly) contributes nothing to the rough figure
-                }
-            }
-        }
-        return total;
-    }
-
    /** Resolve the directory that will hold per-session worktree checkouts. */
    private Path resolveRoot(String repoRoot) {
        if (configuredRoot != null && !configuredRoot.isBlank()) {
@@ -53,14 +53,6 @@ public final class SessionManager implements TurnListener {
    private final int contextCap;
    private final boolean clearAfterTurn;
    private volatile MemberLifecycle memberLifecycle = MemberLifecycle.NONE;
-    /**
-     * CB-586: the repo root the fleet actually works in, remembered the first time a worktree
-     * session is spawned (worktrees are checkouts of it). {@code refs/wip/*} live there, and this
-     * single cached value is what the snapshot retention sweep and the operator-visible census run
-     * against. The daemon is bridged into one project at a time, so "the first worktree's repo" is
-     * the repo; {@code null} until any worktree is spawned, meaning nothing to sweep or measure.
-     */
-    private volatile String fleetRepoRoot;

    /** CB-520: notified with a terminalId on every acquire; no-op until wired. */
    private final List<Consumer<String>> acquireListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
@@ -458,11 +450,6 @@ public final class SessionManager implements TurnListener {
        // The non-worktree path always used this chain; only this branch was missed.
        String repoRoot = worktrees.repoRoot(
                launcher.effectiveCwd(new SpawnRequest(preResolvedProfile, requestedCwd, callerCwd)));
-        if (fleetRepoRoot == null) {
-            // CB-586: remember the repo whose worktrees the fleet spawns — its refs/wip/* are the
-            // snapshot store the retention sweep and the operator census operate on.
-            fleetRepoRoot = repoRoot;
-        }
        String branch = "worker/" + slug(wt.ticketSlug()) + "-" + nonce();
        String path = null;
        PeerHandle handle;
@@ -753,26 +740,6 @@ public final class SessionManager implements TurnListener {
        return registry.size();
    }

-    /**
-     * CB-586: the operator-visible census of {@code refs/wip/*} in the repo the fleet works in —
-     * how many snapshot refs exist and roughly what they cost. Empty (no repo known) until at
-     * least one worktree session has been spawned, exactly so a fleet that has never snapshotted
-     * anything surfaces nothing new, as it did before CB-586.
-     */
-    public Optional<Worktrees.WipRefStats> wipRefs() {
-        String repo = fleetRepoRoot;
-        return repo == null ? Optional.empty() : Optional.of(worktrees.wipRefs(repo));
-    }
-
-    /**
-     * CB-586: run the snapshot retention sweep in the fleet's repo (a no-op until a worktree has
-     * been spawned, which establishes the repo). Returns how many {@code refs/wip/*} it deleted.
-     */
-    public int sweepWipRefs(long minAgeMillis) {
-        String repo = fleetRepoRoot;
-        return repo == null ? 0 : worktrees.pruneWipRefs(repo, minAgeMillis);
-    }
-
    /**
     * The registered session owning {@code terminalId}, or {@code null} if none does.
     *
@@ -14,20 +14,12 @@ public final class SessionReaper {

    private static final Logger log = LoggerFactory.getLogger(SessionReaper.class);
    private static final long DEFAULT_INTERVAL_MILLIS = 5000;
-    /** CB-586: the refs/wip age floor — never sweep a snapshot younger than 24h (the CB-586 rule). */
-    private static final long WIP_MIN_AGE_MILLIS = TimeUnit.HOURS.toMillis(24);
-    /**
-     * CB-586: how often the retention sweep runs. Given the 24h age floor, running it every few
-     * hours means a ref is dropped within hours of becoming eligible, never within minutes.
-     */
-    private static final long WIP_SWEEP_INTERVAL_NANOS = TimeUnit.HOURS.toNanos(6);

    private final SessionManager sessions;
    private final long idleTtlNanos;
    private final long intervalMillis;
    private volatile boolean running;
    private Thread thread;
-    private volatile long lastWipSweepNanos = Long.MIN_VALUE;

    /** Construct a reaper with the default 5-second polling interval. */
    public SessionReaper(SessionManager sessions, long idleTtlSeconds) {
@@ -57,32 +49,10 @@ public final class SessionReaper {
            } catch (RuntimeException e) {
                log.warn("session reaper iteration failed; continuing", e);
            }
-            maybeSweepWipRefs();
            sleep();
        }
    }

-    /**
-     * CB-586: run the refs/wip retention sweep on a slow cadence (hours, not the per-iteration
-     * millisecond loop). Best-effort — a failure must never take the idle-reap loop down with it.
-     */
-    private void maybeSweepWipRefs() {
-        long now = System.nanoTime();
-        if (now - lastWipSweepNanos < WIP_SWEEP_INTERVAL_NANOS) {
-            return;
-        }
-        try {
-            int deleted = sessions.sweepWipRefs(WIP_MIN_AGE_MILLIS);
-            if (deleted > 0) {
-                log.info("refs/wip retention sweep deleted {} snapshot ref(s) older than 24h whose "
-                        + "content was already reachable from main", deleted);
-            }
-        } catch (RuntimeException e) {
-            log.warn("refs/wip retention sweep failed; continuing", e);
-        }
-        lastWipSweepNanos = now;
-    }
-
    private void sleep() {
        try {
            Thread.sleep(intervalMillis);
@@ -54,48 +54,4 @@ public interface Worktrees {
     *         tolerance — a worktree that is gone holds nothing to snapshot)
     */
    Optional<String> snapshot(String worktreePath, String branch, String message);
-
-    /**
-     * CB-586: how many {@code refs/wip/*} snapshot refs exist in {@code repoRoot} and roughly what
-     * they cost. This is the operator-visible surface for the snapshot growth CB-578 stage C left
-     * behind — counts of refs alone hide that each one pins a whole tree for {@code git gc}.
-     *
-     * @param repoRoot the repository to scan
-     * @return count of snapshot refs, and {@code costBytes} = the approximate total working-tree
-     *         size of every snapshot's committed content (summed per ref, so shared objects are
-     *         counted once per ref that carries them)
-     */
-    WipRefStats wipRefs(String repoRoot);
-
-    /**
-     * CB-586: run the {@code refs/wip/*} retention sweep and return how many refs it deleted.
-     *
-     * <p>The retention rule is <em>reachability plus an age floor</em>. A snapshot ref is deleted
-     * only when <strong>both</strong> hold:
-     * <ol>
-     *   <li>its commit's <em>tree content</em> is already reachable from {@code main} — the work
-     *       the snapshot preserved has been recovered, so dropping the ref loses nothing; and</li>
-     *   <li>the ref is older than {@code minAgeMillis} — a very recent snapshot is never swept
-     *       while a lead may still be looking at it.</li>
-     * </ol>
-     *
-     * <p>Reachability is the safety property. A snapshot exists precisely because the work was not
-     * committed anywhere else, so a snapshot whose content is <em>not</em> reachable from
-     * {@code main} is the <strong>last copy</strong> of a worker's work and must never be deleted
-     * automatically — that is the failure CB-576 and CB-578 stage C were built to stop. Age alone
-     * must never drive a deletion, because age-based sweeping is exactly how the last copy gets
-     * destroyed. (Both numbers and the rule are CB-586's decision; this method only implements it.)
-     *
-     * <p>Every deletion logs the ref name and the commit sha, so an operator who finds they lost
-     * the wrong thing can still recover it from git's reflog.
-     *
-     * @param repoRoot     the repository whose {@code refs/wip/*} to sweep
-     * @param minAgeMillis the age floor; a ref younger than this is never touched
-     * @return the number of snapshot refs deleted
-     */
-    int pruneWipRefs(String repoRoot, long minAgeMillis);
-
-    /** CB-586: the operator-visible census of {@code refs/wip/*} in one repository. */
-    record WipRefStats(int count, long costBytes) {
-    }
 }
@@ -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)
@@ -27,15 +27,11 @@ public final class FakeWorktrees implements Worktrees {
    public record SnapshotCall(String worktreePath, String branch, String message) {
    }

-    public record PruneCall(String repoRoot, long minAgeMillis) {
-    }
-
    private final List<AddCall> addCalls = new CopyOnWriteArrayList<>();
    private final List<RemoveCall> removeCalls = new CopyOnWriteArrayList<>();
    private final List<OverlayCall> overlayCalls = new CopyOnWriteArrayList<>();
    private final List<RepoRootCall> repoRootCalls = new CopyOnWriteArrayList<>();
    private final List<SnapshotCall> snapshotCalls = new CopyOnWriteArrayList<>();
-    private final List<PruneCall> pruneCalls = new CopyOnWriteArrayList<>();
    private final Set<String> existingPaths = ConcurrentHashMap.newKeySet();
    private final Set<String> trackedPaths = ConcurrentHashMap.newKeySet();
    private final AtomicLong snapshotSeq = new AtomicLong();
@@ -44,8 +40,6 @@ public final class FakeWorktrees implements Worktrees {
    private volatile boolean dirty = false;
    private volatile String repoRoot = "/repo";
    private volatile String prefix = "/worktrees";
-    private volatile WipRefStats wipRefs = new WipRefStats(0, 0L);
-    private volatile int pruneResult = 0;

    public FakeWorktrees withRepoRoot(String root) {
        this.repoRoot = root;
@@ -88,18 +82,6 @@ public final class FakeWorktrees implements Worktrees {
        return this;
    }

-    /** Configure the value returned by {@link #wipRefs}. */
-    public FakeWorktrees withWipRefs(WipRefStats stats) {
-        this.wipRefs = stats;
-        return this;
-    }
-
-    /** Configure the value returned by {@link #pruneWipRefs}. */
-    public FakeWorktrees withPruneResult(int deleted) {
-        this.pruneResult = deleted;
-        return this;
-    }
-
    @Override
    public String add(String repoRoot, String branch, String baseRef) {
        addCalls.add(new AddCall(repoRoot, branch, baseRef));
@@ -153,17 +135,6 @@ public final class FakeWorktrees implements Worktrees {
        return Optional.of("wip" + snapshotSeq.incrementAndGet());
    }

-    @Override
-    public WipRefStats wipRefs(String repoRoot) {
-        return wipRefs;
-    }
-
-    @Override
-    public int pruneWipRefs(String repoRoot, long minAgeMillis) {
-        pruneCalls.add(new PruneCall(repoRoot, minAgeMillis));
-        return pruneResult;
-    }
-
    public List<AddCall> addCalls() {
        return List.copyOf(addCalls);
    }
@@ -196,10 +167,6 @@ public final class FakeWorktrees implements Worktrees {
        return List.copyOf(snapshotCalls);
    }

-    public List<PruneCall> pruneCalls() {
-        return List.copyOf(pruneCalls);
-    }
-
    public SnapshotCall lastSnapshot() {
        return snapshotCalls.isEmpty() ? null : snapshotCalls.getLast();
    }
@@ -3,7 +3,6 @@ package dev.ltms.bridged.session;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;

-import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashSet;
@@ -139,53 +138,6 @@ class GitWorktreesTest {
        return out;
    }

-    /** Write {@code content} as a blob into the object database; returns its sha. */
-    private static String blobOf(Path cwd, String content) throws Exception {
-        Process p = new ProcessBuilder("git", "-C", cwd.toString(), "hash-object", "-w", "--stdin")
-                .redirectErrorStream(true).start();
-        p.getOutputStream().write(content.getBytes(StandardCharsets.UTF_8));
-        p.getOutputStream().close();
-        String out = new String(p.getInputStream().readAllBytes()).trim();
-        assertTrue(p.waitFor(30, TimeUnit.SECONDS), "git hash-object timed out");
-        assertEquals(0, p.exitValue(), "git hash-object failed:\n" + out);
-        return out;
-    }
-
-    /** Build a single-file tree object from {@code blob}; returns the tree's sha. */
-    private static String treeOf(Path cwd, String path, String blob) throws Exception {
-        Process p = new ProcessBuilder("git", "-C", cwd.toString(), "mktree")
-                .redirectErrorStream(true).start();
-        p.getOutputStream().write(("100644 blob " + blob + "\t" + path + "\n").getBytes(StandardCharsets.UTF_8));
-        p.getOutputStream().close();
-        String out = new String(p.getInputStream().readAllBytes()).trim();
-        assertTrue(p.waitFor(30, TimeUnit.SECONDS), "git mktree timed out");
-        assertEquals(0, p.exitValue(), "git mktree failed:\n" + out);
-        return out;
-    }
-
-    /** {@code git commit-tree} rooted at {@code tree} with a chosen committer date; returns the sha. */
-    private static String commitTree(Path cwd, String tree, String parent, String committerDate,
-                                     String message) throws Exception {
-        ProcessBuilder pb = new ProcessBuilder("git", "-C", cwd.toString(), "commit-tree",
-                tree, "-p", parent, "-m", message);
-        pb.environment().put("GIT_COMMITTER_DATE", committerDate);
-        Process p = pb.redirectErrorStream(true).start();
-        String out = new String(p.getInputStream().readAllBytes()).trim();
-        assertTrue(p.waitFor(30, TimeUnit.SECONDS), "git commit-tree timed out");
-        assertEquals(0, p.exitValue(), "git commit-tree failed:\n" + out);
-        return out;
-    }
-
-    /** {@code git update-ref <ref> <sha>} — create the snapshot ref directly. */
-    private static void updateRef(Path cwd, String ref, String sha) throws Exception {
-        git(cwd, "update-ref", ref, sha);
-    }
-
-    /** True when {@code ref} exists in the repo (for-each-ref on a missing ref is empty, not an error). */
-    private static boolean refExists(Path cwd, String ref) throws Exception {
-        return !forEachRef(cwd, ref).trim().isEmpty();
-    }
-
    /**
     * The heart of CB-525: a provisioned worktree must not inherit the primary's MCP servers. Without
     * the isolation step the checked-out {@code .mcp.json} carries them in, and a worker navigating
@@ -522,104 +474,4 @@ class GitWorktreesTest {
        assertThrows(WorktreeException.class, () -> gitWorktrees.snapshot(wt, branch, "test snapshot"),
                "an unresolvable real index must fail loudly, not silently snapshot from an empty index");
    }
-
-    /**
-     * CB-586, criterion 2. A snapshot whose content is NOT reachable from {@code main} is the last
-     * copy of a worker's work, and must never be deleted automatically — even when it is old and
-     * even when the caller passes a zero age floor. Uses the real snapshot path on a dirty worktree,
-     * so the unreachable tree is exactly the shape CB-576/CB-578 stage C exist to protect.
-     */
-    @Test
-    void anUnreachableSnapshotIsNeverPruned(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
-        String branch = "cb-586-unreachable";
-        String wt = gitWorktrees.add(repo.toString(), branch, "HEAD");
-        Files.writeString(Path.of(wt).resolve("worker-draft.txt"), "work that exists nowhere else\n");
-
-        Optional<String> ref = gitWorktrees.snapshot(wt, branch, "snapshot with unreachable content");
-        assertTrue(ref.isPresent());
-
-        // Age floor 0 makes age a non-issue: only reachability can save it — and it must.
-        assertEquals(0, gitWorktrees.pruneWipRefs(repo.toString(), 0),
-                "the unreachable snapshot is the last copy and must not be pruned");
-        assertTrue(refExists(repo, "refs/wip/" + branch),
-                "an unreachable snapshot must survive the sweep");
-    }
-
-    /**
-     * CB-586, criterion 1 (the reachable half). A snapshot whose tree content IS already reachable
-     * from {@code main} and which is older than the age floor is pure duplication — the work is
-     * recovered — so it must be pruned.
-     */
-    @Test
-    void aReachableSnapshotOlderThanTheFloorIsPruned(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
-
-        // A snapshot whose tree is exactly main's current tree: fully reachable from main.
-        String mainTree = revParse(repo, "main^{tree}");
-        String old = commitTree(repo, mainTree, revParse(repo, "HEAD"), "2020-01-01T00:00:00", "snapshot");
-        updateRef(repo, "refs/wip/recovered", old);
-
-        assertEquals(1, gitWorktrees.pruneWipRefs(repo.toString(), TimeUnit.HOURS.toMillis(24)),
-                "an old, main-reachable snapshot must be pruned");
-        assertFalse(refExists(repo, "refs/wip/recovered"),
-                "the reachable snapshot's ref must be gone after the sweep");
-    }
-
-    /**
-     * CB-586, the age floor. A snapshot whose content IS reachable from {@code main} but which is
-     * younger than the age floor must not be swept — a lead may still be looking at it.
-     */
-    @Test
-    void aReachableButRecentSnapshotIsNotPruned(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
-
-        // Reachable from main, but committed "now" — a fresh snapshot. The 24h floor must protect it.
-        String mainTree = revParse(repo, "main^{tree}");
-        String fresh = commitTree(repo, mainTree, revParse(repo, "HEAD"),
-                "2038-01-01T00:00:00", "snapshot just taken");
-        updateRef(repo, "refs/wip/fresh", fresh);
-
-        assertEquals(0, gitWorktrees.pruneWipRefs(repo.toString(), TimeUnit.HOURS.toMillis(24)),
-                "a recent snapshot must be kept even when reachable");
-        assertTrue(refExists(repo, "refs/wip/fresh"),
-                "the recent reachable snapshot must survive the sweep");
-    }
-
-    /**
-     * CB-586, criterion 5. A fleet that has never snapshotted anything has no {@code refs/wip/*},
-     * so a sweep is a no-op and the census reports none — identical to before CB-586 existed.
-     */
-    @Test
-    void aFleetWithNoSnapshotsPrunesNothingAndReportsNothing(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
-
-        assertEquals(0, gitWorktrees.pruneWipRefs(repo.toString(), 0),
-                "no snapshot refs means nothing to prune");
-        Worktrees.WipRefStats stats = gitWorktrees.wipRefs(repo.toString());
-        assertEquals(0, stats.count(), "a never-snapshotted fleet has zero refs/wip refs");
-        assertEquals(0L, stats.costBytes(), "a never-snapshotted fleet costs zero bytes");
-    }
-
-    /** CB-586, criterion 4: the census reports how many refs exist and roughly what they cost. */
-    @Test
-    void wipRefsReportsCountAndCost(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
-
-        String blob = blobOf(repo, "a recoverable snapshot's worth of content");
-        String tree = treeOf(repo, "snapshot.txt", blob);
-        updateRef(repo, "refs/wip/one", commitTree(repo, tree, revParse(repo, "HEAD"),
-                "2020-01-01T00:00:00", "snapshot"));
-        updateRef(repo, "refs/wip/two", commitTree(repo, tree, revParse(repo, "HEAD"),
-                "2020-01-02T00:00:00", "snapshot"));
-
-        Worktrees.WipRefStats stats = gitWorktrees.wipRefs(repo.toString());
-        assertEquals(2, stats.count(), "two snapshot refs are reported");
-        assertTrue(stats.costBytes() > 0, "the cost of the snapshots is a positive byte count");
-    }
 }
@@ -138,16 +138,6 @@ class SessionManagerTest {
            return java.util.Optional.of("wip" + snapshotSeq.incrementAndGet());
        }

-        @Override
-        public WipRefStats wipRefs(String repoRoot) {
-            return new WipRefStats(0, 0L);
-        }
-
-        @Override
-        public int pruneWipRefs(String repoRoot, long minAgeMillis) {
-            return 0;
-        }
-
        List<String> removeCalls() {
            return List.copyOf(removeCalls);
        }
@@ -444,37 +444,4 @@ class WorktreeSessionManagerTest {
                "a failed snapshot leaves no ref to report");
    }

-    /**
-     * CB-586, criteria 4 and 1. Once a worktree session is spawned the repo is known, so the
-     * operator census and the retention sweep delegate to that repo's {@code refs/wip/*}.
-     */
-    @Test
-    void wipRefsAndSweepDelegateToTheFleetRepoOnceKnown() {
-        FakeHerdr herdr = new FakeHerdr();
-        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
-                .withWipRefs(new Worktrees.WipRefStats(3, 42L)).withPruneResult(2);
-        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
-
-        assertTrue(sessions.wipRefs().isEmpty(),
-                "no worktree spawned yet means no repo is known and nothing to report");
-        assertEquals(0, sessions.sweepWipRefs(TimeUnit.HOURS.toMillis(24)),
-                "no worktree spawned yet means the sweep is a no-op");
-
-        sessions.acquire("ltms-local", null, "/caller/proj", null,
-                new WorktreeRequest("cb-586", null));
-
-        Worktrees.WipRefStats stats = sessions.wipRefs().orElseThrow();
-        assertEquals(3, stats.count(), "the census comes from the fleet repo");
-        assertEquals(42L, stats.costBytes(), "the cost comes from the fleet repo");
-        assertEquals(2, sessions.sweepWipRefs(TimeUnit.HOURS.toMillis(24)),
-                "the sweep runs against the fleet repo");
-
-        List<FakeWorktrees.PruneCall> prunes = worktrees.pruneCalls();
-        assertEquals(1, prunes.size(), "the no-op short-circuits before reaching the seam, so only "
-                + "the repo-known sweep issues a call");
-        assertEquals("/repo", prunes.getFirst().repoRoot(), "the sweep targets the fleet repo");
-        assertEquals(TimeUnit.HOURS.toMillis(24), prunes.getFirst().minAgeMillis(),
-                "the caller's age floor is passed through");
-    }
-
 }