#317: refuse an unresolved caller instead of promoting it to primary

ConnectionIdentity.resolve() called pids.pidForLocalPort(remotePort),
which returns -1 both on a real failure and (silently, no log line)
when lsof just finds no matching process. terminalForPid(-1) then
matches no pane, so CallerResolver's loopback-trust fallback could not
tell that caller apart from a genuine primary and handed it
Principal.primary(...) — granting SPAWN, STOP, SEND and DRAIN to a
worker whose PID lookup failed. This is the escalation PaneLocator's
own javadoc already names; CB-161's ancestry walk only helps once a
candidate pid exists, and a failed lookup has none.

Fix: ConnectionIdentity.Caller gets a resolved() predicate (pid > 0),
centralised next to the -1 sentinel it tests for the same reason
isLoopback() is centralised (fleetd #305: two independent copies of
one rule already drifted once). CallerResolver's loopback-trust
fallback now requires c.resolved() before granting PRIMARY; an
unresolved caller gets Principal.anonymous() — the same already-tested
"authenticated as nothing" outcome used everywhere else in that
method, so the refusal is a clean, named, unsurprising result rather
than something that looks like a bug.

Also logs the previously-silent "lsof ran clean, found no match" case
in LsofPeerPidLookup at DEBUG, since that (not a slow lsof — the
waitFor result was already discarded) is the likelier real trigger.

loopbackTrustTreatsANonWorkerLoopbackCallerAsThePrimary is untouched
and still green: a real pid that owns no pane (the actual primary) is
still resolved() and still PRIMARY. Token mode is unaffected — it
never consults c.pid() at all.

Mutation-tested: reverting only the CallerResolver.java guard
reproduces the escalation exactly (aFailedPeerPidLookupIsRefusedNotPromotedToPrimary
fails with "expected: <ANONYMOUS> but was: <PRIMARY>").

fleetd/src/main/java/dev/ltms/fleet/auth/CallerResolver.java

@@ -236,7 +236,15 @@ public final class CallerResolver {
         // loopback-trust: same-host callers that are not workers are the primary. A non-loopback
         // caller is anonymous even here — and startup refuses that combination anyway
         // (FleetConfig.validateAuthExposure), so this is defence in depth, not the control.
-        return isLoopback(remoteAddr) ? Principal.primary(c.pid()) : Principal.anonymous();
+        //
+        // fleetd #317: "not a worker" must not be conflated with "identity unresolved". The real
+        // primary is a real process — its pid resolves (c.resolved()), it just owns no herdr pane.
+        // A caller whose peer-PID lookup failed (LsofPeerPidLookup's -1 sentinel — on any failure,
+        // silently including "lsof found no match") has no such pid, and PaneLocator's own javadoc
+        // already names what happens if that case is handed the primary role: a worker→primary
+        // escalation. So an unresolved caller is refused (ANONYMOUS — the same clean, already-tested
+        // "authenticated as nothing" outcome used everywhere else in this method), never promoted.
+        return isLoopback(remoteAddr) && c.resolved() ? Principal.primary(c.pid()) : Principal.anonymous();
     }
 
     private boolean presentedTokenMatches(String authorizationHeader) {

fleetd/src/main/java/dev/ltms/fleet/mcp/ConnectionIdentity.java

@@ -36,6 +36,25 @@ public final class ConnectionIdentity {
      * primary / an off-host client) and its {@code pid} (or {@code -1} if not resolvable).
      */
     public record Caller(String terminal, long pid) {
+
+        /**
+         * Whether the OS peer-PID lookup actually succeeded — {@code false} means {@code pid} is
+         * the {@code -1} sentinel, not a real process id, so this caller's identity could not be
+         * established at all. That is a different fact from a real pid that simply owns no worker
+         * pane (the primary's own connection): the primary is {@code resolved()} and has a
+         * {@code null terminal}; an unresolvable caller is {@code !resolved()} and also has a
+         * {@code null terminal}. The two look identical through {@link #terminal} alone, which is
+         * exactly how fleetd #317 happened — a failed {@code lsof} lookup and a genuine primary both
+         * fell through to {@code Principal.primary(...)}.
+         *
+         * <p>Centralised here, next to the sentinel it tests, for the same reason
+         * {@link ConnectionIdentity#isLoopback} is centralised rather than left for each caller to
+         * reimplement: a raw {@code pid > 0} check duplicated at every call site is precisely the
+         * "one rule, two copies" shape that let #305 drift.
+         */
+        public boolean resolved() {
+            return pid > 0;
+        }
     }
 
     /** Resolve the caller's terminal and PID from one peer-PID lookup. */

fleetd/src/main/java/dev/ltms/fleet/mcp/LsofPeerPidLookup.java

@@ -41,6 +41,14 @@ public final class LsofPeerPidLookup implements PeerPidLookup {
             if (!p.waitFor(2, TimeUnit.SECONDS)) {
                 p.destroyForcibly();
             }
+            if (found < 0) {
+                // fleetd #317: this is the silent path — lsof ran clean and simply reported no
+                // matching process (e.g. queried before the OS socket table settles). Previously
+                // this logged nothing at all, which is exactly why the escalation went unnoticed;
+                // the exception path below already logs. A caller now refused because of this is
+                // still refused (never promoted) — this line only makes the refusal diagnosable.
+                log.debug("lsof peer-pid lookup for port {} found no matching process", port);
+            }
             return found;
         } catch (Exception e) {
             log.debug("lsof peer-pid lookup for port {} failed: {}", port, e.getMessage());

fleetd/src/main/java/dev/ltms/fleet/msg/AmqpReplyInbox.java

@@ -22,7 +22,6 @@ import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * AMQP-backed {@link ReplyInbox} (CB-307 Stage 2): genuine cross-restart durability behind the same
@@ -94,23 +93,8 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     private final Channel channel;
     /** All channel operations (publish/declare/ack/cancel) serialize on this — a Channel is not thread-safe. */
     private final Object channelLock = new Object();
-    /**
-     * target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself.
-     *
-     * <p><strong>CB-318 tombstone.</strong> The value {@link #RELEASED} is a reserved sentinel: it
-     * marks a target whose {@link #release} has already run, so {@link #deliverCallback} can tell a
-     * delivery landing after release() apart from a fresh target it has never seen. See both methods'
-     * javadoc for why a plain {@code held.remove(target)} is not enough.
-     */
+    /** target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself. */
     private final ConcurrentHashMap<String, LinkedHashMap<String, Held>> held = new ConcurrentHashMap<>();
-
-    /**
-     * CB-318 sentinel stored in {@link #held} for a target whose {@link #release} has already run.
-     * Never mutated — every read site compares it by reference ({@code ==}) before touching it as a
-     * map, because it is a single object shared across every released target and calling a mutator on
-     * it would corrupt state for all of them.
-     */
-    private static final LinkedHashMap<String, Held> RELEASED = new LinkedHashMap<>();
     /** Targets whose queue is declared and consumer is running, mapped to their broker consumer tag. */
     private final ConcurrentHashMap<String, String> consumerTags = new ConcurrentHashMap<>();
 
@@ -217,12 +201,6 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
                 channel.queueDeclare(queue, true, false, false, null); // durable, non-exclusive, keep on idle
                 String tag = channel.basicConsume(queue, false, deliverCallback(target), _ -> { });
                 consumerTags.put(target, tag);
-                // CB-318: drop a stale RELEASED tombstone from a prior ownership of this same target
-                // string, so a delivery under this fresh consumer is held normally instead of being
-                // nacked forever by deliverCallback's RELEASED check. Safe to do here, still under
-                // channelLock: no delivery for the consumer tag just registered above can reach
-                // deliverCallback before this basicConsume call returns.
-                held.remove(target, RELEASED);
                 log.debug("AMQP inbox owns queue {} for target {}", queue, target);
             } catch (IOException e) {
                 throw new IllegalStateException("cannot own queue " + queue, e);
@@ -265,32 +243,6 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
      * later connection drop even though this release did not manage to requeue it immediately. A
      * failed {@code basicCancel} still throws, unchanged from before this fix — that failure means
      * the consumer may still be attached, so best-effort requeue is not attempted underneath it.
-     *
-     * <p><strong>CB-318: {@code held.remove(target)} alone leaves a second window open.</strong> The
-     * bullet above already explains why cancelling first does not save a tag from going stale — but
-     * that only accounts for a delivery landing before this method starts touching {@link #held}.
-     * {@code basicCancel} stops <em>new</em> dispatches; it does not flush one already handed to the
-     * consumer work pool. So a delivery can still land on that pool's thread and reach
-     * {@link #deliverCallback} at any point during, or after, this method's body — and a plain
-     * {@code held.remove(target)} does nothing to stop it: {@code deliverCallback}'s
-     * {@code computeIfAbsent} finds the key gone and happily creates a brand-new map under it, which
-     * this method — already past its {@code remove} — never looks at again. That entry then sits
-     * delivered-but-unacked on {@link #channel} until the whole inbox closes: never requeued, never
-     * redelivered, and {@link #peek} is never called again for a target nothing owns any more.
-     *
-     * <p>The fix is {@link #held}{@code .compute(target, ...)} instead of {@code remove}: it takes
-     * whatever was held (to nack, same as before) and, in the same atomic step, leaves the
-     * {@link #RELEASED} tombstone behind instead of an absent key. {@code computeIfAbsent} and
-     * {@code compute} calls for the same key are mutually exclusive in {@link ConcurrentHashMap} —
-     * whichever of this call and a concurrent {@code deliverCallback} runs first is fully visible to
-     * the other, with no gap between them. So a delivery that loses the race sees a real map here and
-     * gets nacked by the loop below, same as always; a delivery that wins the race (runs first) is
-     * itself nacked by that same loop, once it settles into {@code held}. A delivery that arrives once
-     * this method has stored {@link #RELEASED} finds it via {@code computeIfAbsent} and refuses itself
-     * — see {@link #deliverCallback}. Either way nothing is silently retained forever, satisfying the
-     * ticket's invariant against dropping a message. This closes the window rather than merely
-     * narrowing it — correctness does not depend on how much time elapses between the swap and this
-     * method returning.
      */
     @Override
     public void release(String target) {
@@ -303,13 +255,8 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
                     throw new IllegalStateException("cannot cancel consumer for " + target, e);
                 }
             }
-            AtomicReference<LinkedHashMap<String, Held>> previouslyHeld = new AtomicReference<>();
-            held.compute(target, (_, v) -> {
-                previouslyHeld.set(v);
-                return RELEASED;
-            });
-            var perTarget = previouslyHeld.get();
-            if (perTarget != null && perTarget != RELEASED) {
+            var perTarget = held.remove(target);
+            if (perTarget != null) {
                 synchronized (perTarget) {
                     for (Held h : perTarget.values()) {
                         try {
@@ -379,7 +326,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     @Override
     public List<InboxMessage> peek(String target) {
         var perTarget = held.get(target);
-        if (perTarget == null || perTarget == RELEASED) {
+        if (perTarget == null) {
             return List.of();
         }
         synchronized (perTarget) {
@@ -390,7 +337,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     @Override
     public void ack(String target, String msgId) {
         var perTarget = held.get(target);
-        if (perTarget == null || perTarget == RELEASED) {
+        if (perTarget == null) {
             return;
         }
         Held h;
@@ -423,20 +370,6 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
             }
             String content = new String(delivery.getBody(), StandardCharsets.UTF_8);
             var perTarget = held.computeIfAbsent(target, _ -> new LinkedHashMap<>());
-            if (perTarget == RELEASED) {
-                // CB-318: release() already ran for this target and left the RELEASED tombstone in
-                // held (see release()'s javadoc) — computeIfAbsent() is guaranteed to see it rather
-                // than recreate a fresh map, because ConcurrentHashMap serializes compute/
-                // computeIfAbsent calls for the same key against each other. Refuse the delivery
-                // instead of holding it somewhere release() will never look at again: requeue it, the
-                // same way release() nacks its own held entries, so a later owner (or a connection
-                // drop) can still recover it. This does not need channelLock across a broker round
-                // trip — basicNack, like the duplicate-ack case just below, does not wait for one.
-                synchronized (channelLock) {
-                    channel.basicNack(tag, false, true);
-                }
-                return;
-            }
             boolean duplicate;
             synchronized (perTarget) {
                 if (perTarget.containsKey(msgId)) {

fleetd/src/test/java/dev/ltms/fleet/auth/CallerResolverTest.java

@@ -103,6 +103,54 @@ class CallerResolverTest {
         assertEquals(Role.PRIMARY, p.role(), "the historical behaviour, now an explicit choice");
     }
 
+    // ── fleetd #317: an unresolvable caller must never be promoted to the primary ──────────────────
+    // #305 closed the trigger where a resolved pid matched no pane *and* had no ancestry walk to
+    // save it. This is the other trigger PaneLocator's javadoc names: the pid never resolves at
+    // all — LsofPeerPidLookup returns -1 on any failure, including (silently) "lsof found no
+    // match" — so there is no candidate pid for an ancestry walk to even attempt.
+
+    /**
+     * The failing-without-the-fix case. Before #317's fix, {@code c.terminal() == null} was the
+     * only test in the loopback-trust fallback, and an unresolved pid produces exactly that same
+     * {@code null} terminal as a genuine primary — so this caller was handed
+     * {@code Principal.primary(...)}, a real worker's failed lookup becoming indistinguishable from
+     * the lead.
+     */
+    @Test
+    void aFailedPeerPidLookupIsRefusedNotPromotedToPrimary() {
+        ConnectionIdentity unresolved = new ConnectionIdentity(new PaneLocator(herdr), _ -> -1);
+        Principal p = new CallerResolver(unresolved).resolve("127.0.0.1", 55555, null);
+
+        assertEquals(Role.ANONYMOUS, p.role(),
+                "an unresolvable caller must never be silently promoted to the primary");
+    }
+
+    /**
+     * The companion invariant #317 must not break: a caller whose lookup genuinely succeeded, and
+     * who simply owns no herdr pane — the real primary's own connection — is still the primary.
+     * This is {@link #loopbackTrustTreatsANonWorkerLoopbackCallerAsThePrimary} pinned again here,
+     * named for #317 and placed next to the test it must be distinguished from: same {@code null}
+     * terminal, opposite verdict, because {@code Caller.resolved()} tells them apart.
+     */
+    @Test
+    void aRealPidThatOwnsNoPaneIsStillThePrimaryNotRefused() {
+        Principal p = new CallerResolver(nonWorkerIdentity()).resolve("127.0.0.1", 55555, null);
+
+        assertEquals(Role.PRIMARY, p.role());
+    }
+
+    /** #317 point 4: token mode never consults {@code c.pid()}, so a failed lookup must not change it. */
+    @Test
+    void tokenModeIsUndisturbedByAnUnresolvedLookup() {
+        ConnectionIdentity unresolved = new ConnectionIdentity(new PaneLocator(herdr), _ -> -1);
+        CallerResolver r = new CallerResolver(unresolved, true, "s3cret");
+
+        assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 55555, null).role(),
+                "no credential is still just ANONYMOUS, as before #317 — unchanged by the lookup failing");
+        assertEquals(Role.PRIMARY, r.resolve("127.0.0.1", 55555, "Bearer s3cret").role(),
+                "a valid token still authenticates the primary even though the peer-pid lookup failed");
+    }
+
     @Test
     void tokenModeRefusesANonWorkerCallerThatPresentsNoToken() {
         Principal p = new CallerResolver(nonWorkerIdentity(), true, "s3cret")

fleetd/src/test/java/dev/ltms/fleet/mcp/ConnectionIdentityTest.java

@@ -43,6 +43,22 @@ class ConnectionIdentityTest {
         assertNull(with(_ -> 999_999).callerTerminal("127.0.0.1", 55555));
     }
 
+    @Test
+    void callerIsUnresolvedWhenThePeerPidLookupFails() {
+        // fleetd #317: LsofPeerPidLookup returns -1 on any failure — a fork error, or (silently)
+        // simply no matching lsof line. Caller.resolved() is the one place that sentinel is tested.
+        ConnectionIdentity.Caller c = with(_ -> -1).resolve("127.0.0.1", 55555);
+        assertFalse(c.resolved(), "a -1 pid means the lookup failed, not that this pid owns no pane");
+    }
+
+    @Test
+    void callerIsResolvedWhenThePidIsRealEvenThoughItOwnsNoPane() {
+        // The primary's own connection: a real, lsof-found pid that just isn't a worker pane. This
+        // must read as "resolved" — the distinction #317 turns on.
+        ConnectionIdentity.Caller c = with(_ -> 999_999).resolve("127.0.0.1", 55555);
+        assertTrue(c.resolved());
+    }
+
     @Test
     void resolvesTheCallersPidAndCwd() {
         // CB-112: the primary maps to no pane, but its PID and cwd are still readable.

fleetd/src/test/java/dev/ltms/fleet/msg/AmqpReplyInboxReleaseRaceTest.java

@@ -1,264 +0,0 @@
-package dev.ltms.fleet.msg;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.DeliverCallback;
-import com.rabbitmq.client.Delivery;
-import com.rabbitmq.client.Envelope;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
-
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Proxy;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * CB-318: a delivery landing on the consumer work-pool thread <em>after</em>
- * {@link AmqpReplyInbox#release} has already swapped the target's {@code held} entry for its
- * tombstone, but <em>before</em> {@code release()} itself returns, must be nacked-with-requeue —
- * never silently retained in a fresh map {@code release()} has already stopped looking at.
- *
- * <p><strong>This forces the actual interleaving, not a sequence of calls.</strong> {@code release()}
- * runs on its own thread and is made to block <em>inside</em> its nack loop, via a fake
- * {@link Channel} whose {@code basicNack} blocks on its first invocation. That block is only
- * reachable after {@code release()}'s {@code held.compute(...)} has already swapped in the
- * {@code RELEASED} tombstone (the compute call happens strictly before the loop that calls
- * {@code basicNack}), so observing it is direct, ordering-guaranteed proof that the tombstone is in
- * place and {@code release()} has not yet returned — still holding {@code channelLock} — when a
- * second thread fires {@code own()}'s captured {@link DeliverCallback} for a brand-new message on the
- * same target. No mocking library is on the classpath, so the fake broker is a {@link Proxy}, the
- * same pattern {@code AmqpReplyInboxRecoveryRaceTest} already uses.
- *
- * <p><strong>What this does and does not prove.</strong> It proves that a delivery whose
- * {@code computeIfAbsent} call is ordered strictly after {@code release()}'s tombstone swap — while
- * {@code release()} is still running — is nacked-with-requeue rather than silently parked forever.
- * It does not drive a real broker: {@code basicNack} here is a recorded call on a fake channel, not a
- * verified requeue-and-redeliver. That half of the contract (a nacked-with-requeue delivery really
- * does come back to a later owner) is already covered against a real broker by
- * {@code AmqpReplyInboxContractTest.releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery}, which
- * this test does not duplicate.
- */
-class AmqpReplyInboxReleaseRaceTest {
-
-    @Test
-    @Timeout(15)
-    void deliveryArrivingWhileReleaseIsStillRunningIsNackedNotStranded() throws Exception {
-        String target = "worker-release-race";
-
-        List<long[]> nacks = new CopyOnWriteArrayList<>();   // {deliveryTag, requeue(1/0)}
-        List<Long> acks = new CopyOnWriteArrayList<>();
-        AtomicReference<DeliverCallback> deliverCallback = new AtomicReference<>();
-        CountDownLatch nackStarted = new CountDownLatch(1);
-        CountDownLatch releaseMayFinishNack = new CountDownLatch(1);
-        AtomicInteger nackCallCount = new AtomicInteger();
-
-        Channel consumeChannel = fakeConsumeChannel(deliverCallback, nacks, acks, nackCallCount,
-                nackStarted, releaseMayFinishNack);
-        Channel publishChannel = fakeInertChannel();
-        Connection connection = fakeConnection(consumeChannel, publishChannel);
-
-        AmqpReplyInbox inbox = new AmqpReplyInbox(connection, AmqpReplyInbox.DEFAULT_PREFETCH);
-        inbox.own(target);
-        assertTrue(deliverCallback.get() != null, "own() must have registered a DeliverCallback");
-
-        // Seed one already-held delivery (m0) so release()'s nack loop has something to iterate, and
-        // therefore somewhere to block, before it can return.
-        deliverCallback.get().handle("ctag", delivery(1L, "m0", "first"));
-
-        AtomicReference<Throwable> releaseError = new AtomicReference<>();
-        Thread releaseThread = new Thread(() -> {
-            try {
-                inbox.release(target);
-            } catch (Throwable t) {
-                releaseError.set(t);
-            }
-        }, "release-under-test");
-        releaseThread.start();
-
-        // This latch only fires from inside the fake channel's basicNack — i.e. from inside
-        // release()'s nack loop, which release()'s code only reaches AFTER held.compute(...) has
-        // already swapped in RELEASED. Waiting for it is direct proof the swap has happened and
-        // release() has not yet returned (it is stuck mid-loop, still holding channelLock).
-        assertTrue(nackStarted.await(10, TimeUnit.SECONDS),
-                "release() never reached its nack loop — it may not have started");
-
-        // The exact interleaving CB-318 describes: a delivery for a NEW message on the same target
-        // lands on the "consumer work-pool thread" (this second thread) while release() is still
-        // running. With the pre-fix code (a bare held.remove(target)) this created a brand-new map
-        // under computeIfAbsent that release() — already past its remove — never looks at again.
-        AtomicReference<Throwable> deliveryError = new AtomicReference<>();
-        Thread deliveryThread = new Thread(() -> {
-            try {
-                deliverCallback.get().handle("ctag", delivery(2L, "m1", "second"));
-            } catch (Throwable t) {
-                deliveryError.set(t);
-            }
-        }, "concurrent-delivery");
-        deliveryThread.start();
-
-        // Head start for the delivery thread to reach (and, on the fixed code, block on)
-        // channelLock — release() still holds it at this point, so a correct fix cannot have
-        // resolved m1's nack yet. Purely in-memory work (computeIfAbsent, a reference compare)
-        // separates deliveryThread.start() from that block point, so 300ms is a large margin, not a
-        // tight timing assumption.
-        Thread.sleep(300);
-        assertEquals(1, nackCallCount.get(),
-                "the concurrent delivery must not resolve its nack before release() gives up "
-                        + "channelLock — if this is 2 already, the interleaving below is not being "
-                        + "tested, only a sequential call");
-
-        releaseMayFinishNack.countDown(); // let release() finish nacking m0 and return
-        assertTrue(releaseThread.join(Duration.ofSeconds(10)), "release() did not finish");
-        assertTrue(deliveryThread.join(Duration.ofSeconds(10)), "the concurrent delivery did not finish");
-
-        assertNull(releaseError.get(), "release() threw: " + releaseError.get());
-        assertNull(deliveryError.get(), "the concurrent delivery threw: " + deliveryError.get());
-
-        assertEquals(2, nacks.size(),
-                "both the pre-held m0 and the concurrently-arriving m1 must be nacked, got: "
-                        + nacks.stream().map(n -> "[tag=" + n[0] + " requeue=" + n[1] + "]").toList());
-        assertTrue(nacks.stream().allMatch(n -> n[1] == 1L),
-                "invariant 1 (never drop): every nack must set requeue=true");
-        assertTrue(nacks.stream().anyMatch(n -> n[0] == 1L), "m0's delivery tag must be nacked");
-        assertTrue(nacks.stream().anyMatch(n -> n[0] == 2L),
-                "m1 — delivered while release() was still running, after the tombstone swap — must be "
-                        + "nacked, not silently retained in a map release() will never look at again");
-        assertTrue(acks.isEmpty(), "invariant 1 (never drop): a held reply must never be basicAck'd");
-    }
-
-    private static Delivery delivery(long tag, String msgId, String body) {
-        Envelope envelope = new Envelope(tag, false, "", "irrelevant");
-        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();
-        return new Delivery(envelope, props, body.getBytes(StandardCharsets.UTF_8));
-    }
-
-    /** A {@link Proxy}-backed consume {@link Channel}: blocks the FIRST {@code basicNack} call on
-     *  {@code releaseMayFinishNack}, after signalling {@code nackStarted} — everything else records
-     *  the call and returns a harmless default, matching the style already used by
-     *  {@code AmqpReplyInboxRecoveryRaceTest}. */
-    private static Channel fakeConsumeChannel(AtomicReference<DeliverCallback> deliverCallback,
-                                               List<long[]> nacks, List<Long> acks,
-                                               AtomicInteger nackCallCount,
-                                               CountDownLatch nackStarted,
-                                               CountDownLatch releaseMayFinishNack) {
-        InvocationHandler handler = (proxy, method, args) -> {
-            String name = method.getName();
-            if (name.equals("basicConsume")) {
-                deliverCallback.set((DeliverCallback) args[2]);
-                return "ctag";
-            }
-            if (name.equals("basicNack")) {
-                long tag = (long) args[0];
-                boolean requeue = (boolean) args[2];
-                if (nackCallCount.incrementAndGet() == 1) {
-                    nackStarted.countDown();
-                    if (!releaseMayFinishNack.await(10, TimeUnit.SECONDS)) {
-                        throw new IllegalStateException("test never released the nack latch");
-                    }
-                }
-                nacks.add(new long[] {tag, requeue ? 1L : 0L});
-                return null;
-            }
-            if (name.equals("basicAck")) {
-                acks.add((long) args[0]);
-                return null;
-            }
-            if (name.equals("equals")) {
-                return proxy == args[0];
-            }
-            if (name.equals("hashCode")) {
-                return System.identityHashCode(proxy);
-            }
-            if (name.equals("toString")) {
-                return "FakeConsumeChannel";
-            }
-            return defaultValue(method.getReturnType());
-        };
-        return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
-                new Class<?>[] {Channel.class}, handler);
-    }
-
-    /** A {@link Proxy}-backed {@link Channel} that answers every call with a harmless default — used
-     *  as the publish channel, which this test never actually publishes on. */
-    private static Channel fakeInertChannel() {
-        InvocationHandler handler = (proxy, method, args) -> {
-            String name = method.getName();
-            if (name.equals("equals")) {
-                return proxy == args[0];
-            }
-            if (name.equals("hashCode")) {
-                return System.identityHashCode(proxy);
-            }
-            if (name.equals("toString")) {
-                return "FakeInertChannel";
-            }
-            return defaultValue(method.getReturnType());
-        };
-        return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
-                new Class<?>[] {Channel.class}, handler);
-    }
-
-    /** A {@link Proxy}-backed {@link Connection} handing out {@code first} then {@code second} from
-     *  successive {@code createChannel()} calls, matching {@link AmqpReplyInbox}'s constructor. */
-    private static Connection fakeConnection(Channel first, Channel second) {
-        AtomicInteger calls = new AtomicInteger();
-        InvocationHandler handler = (proxy, method, args) -> {
-            String name = method.getName();
-            if (name.equals("createChannel") && (args == null || args.length == 0)) {
-                return calls.getAndIncrement() == 0 ? first : second;
-            }
-            if (name.equals("equals")) {
-                return proxy == args[0];
-            }
-            if (name.equals("hashCode")) {
-                return System.identityHashCode(proxy);
-            }
-            if (name.equals("toString")) {
-                return "FakeConnection";
-            }
-            return defaultValue(method.getReturnType());
-        };
-        return (Connection) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
-                new Class<?>[] {Connection.class}, handler);
-    }
-
-    private static Object defaultValue(Class<?> type) {
-        if (!type.isPrimitive() || type == void.class) {
-            return null;
-        }
-        if (type == boolean.class) {
-            return Boolean.FALSE;
-        }
-        if (type == long.class) {
-            return 0L;
-        }
-        if (type == short.class) {
-            return (short) 0;
-        }
-        if (type == byte.class) {
-            return (byte) 0;
-        }
-        if (type == char.class) {
-            return (char) 0;
-        }
-        if (type == double.class) {
-            return 0.0d;
-        }
-        if (type == float.class) {
-            return 0.0f;
-        }
-        return 0;
-    }
-}