Classify named AMQP recovery errors

Both connections were already named at newConnection() -- 'fleetd-reply-inbox' and
'fleetd-lead-mailbox' -- and neither name ever reached the log: 0 occurrences in
fleetd.out, and both connections logged under the same thread name
'[AMQP Connection 10.10.20.13:5672]'. So when one of the two died and never came
back, the log could not say which.

AmqpConnectionFailureLogger extends DefaultExceptionHandler and overrides only the
protected log(String, Throwable) sink that every handle* method calls virtually, so
the identity is added without changing any handler action.

My brief caused a defect here and the correction is the interesting part. I told the
worker the client 'currently uses ForgivingExceptionHandler', read off the log line
c.r.c.i.ForgivingExceptionHandler -- which names where the LOGGER FIELD is declared,
not the instance's class. javap on the jar shows ConnectionFactory's constructor does
'new DefaultExceptionHandler', and DefaultExceptionHandler extends StrictExceptionHandler
extends ForgivingExceptionHandler. The first version therefore extended the base and
silently dropped strict channel-closing on four listener/consumer paths. Now pinned by
a type assertion on both factories plus a behavioural test that handleConsumerException
still closes the channel once.

Verified on merge with a mutation the worker did not run: it mutated the parent class,
so I mutated the copied private-static isSocketClosedOrConnectionReset in the DANGEROUS
direction (always true => every failure logs at WARN and vanishes from the redeploy
gate's ERROR count). Caught: 'inbox failure line ==> expected: <ERROR> but was: <WARN>'.

Merged main in first; the auto-merge compiled. 1379 green, unpiped.

fleetd/src/main/java/dev/ltms/fleet/inject/BackendErrorSink.java

@@ -1,10 +1,12 @@
 package dev.ltms.fleet.inject;
 
 /**
- * Notified when {@link CompletionResolver} actually delivers a typed backend-error classification
- * to a waiting send (fleetd#201 / #227) — never on a race that lost. {@link CompletionResolver}
- * calls this only after {@code Rendezvous.resolveFailure} returns {@code true} for that exact
- * waiter, mirroring the win-only race rule {@link ExhaustionSink} already uses.
+ * Notified when {@link CompletionResolver} has a backend-error match at the start of a pane line,
+ * or both a match and its too-fast crash signature, for a waiting send (fleetd#201 / #227). A text
+ * match inside ordinary pane prose can be a member's report about an error, so it fails the send
+ * without notifying this sink.
+ * {@link CompletionResolver} calls this only after {@code Rendezvous.resolveFailure} returns
+ * {@code true} for that exact waiter, mirroring the win-only race rule {@link ExhaustionSink} uses.
  *
  * <p>The public send result is unchanged by this classification — it is still a failed send
  * ({@code Rendezvous.Kind#FAILED}); this sink is the internal seam a later stage (fleetd#201 Unit

fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java

@@ -380,16 +380,19 @@ public final class CompletionResolver implements TurnListener {
                         + "matched the profile's exhausted pattern): {}", target, reason);
                 // CB-578 stage B: only on the resolution that actually won the race — a late
                 // duplicate must never quarantine a credential twice for one refusal.
-                exhaustionSink.onExhausted(target, reason);
+                if (startsWithExhaustion(matchedLine, exhausted)) {
+                    exhaustionSink.onExhausted(target, reason);
+                }
             }
             return;
         }
         // fleetd#164 (part 2) / fleetd#201: a scrape that read cleanly and produced content still
         // isn't a real reply when that content is the backend's own rejection (e.g. an HTTP 400
         // before the worker did any work). Classify it as a failure naming the member, rather than
-        // handing the caller a scrape that reads like a completed answer, and — only on the
-        // resolution that actually wins the race, mirroring the exhaustion sink above — notify the
-        // typed backend-error sink so a later stage can act on repeated failures.
+        // handing the caller a scrape that reads like a completed answer. A text match alone is not
+        // enough to notify the typed backend-error sink: this assistant block can be a member's
+        // normal prose about an error. A line that starts with the error match is stronger evidence;
+        // the too-fast path below also has its crash signature before it records a credential failure.
         String backendError = firstMatchingLine(assistantBlock, backendErrorPatternOrFallback(target));
         if (backendError != null) {
             // Carry the whole scrape, not just the matched line. The pattern is a heuristic: a member
@@ -401,9 +404,9 @@ public final class CompletionResolver implements TurnListener {
             if (rendezvous.resolveFailure(waiter, reason)) {
                 inFlight.remove(target, turn);
                 log.warn("failing send to {} via turn-stall fallback: {}", target, reason);
-                // fleetd#201 Unit 1: only on the resolution that actually won the race — a late
-                // duplicate must never double-count one backend failure.
-                backendErrorSink.onBackendError(target, backendError, reason);
+                if (startsWithBackendError(backendError, backendErrorPatternOrFallback(target))) {
+                    backendErrorSink.onBackendError(target, backendError, reason);
+                }
             }
             return;
         }
@@ -477,7 +480,9 @@ public final class CompletionResolver implements TurnListener {
                         + "usable assistant block; no fleet_reply): {}", target, reason);
                 // CB-578 stage B: only on the resolution that actually won the race — a late
                 // duplicate must never quarantine a credential twice for one refusal.
-                exhaustionSink.onExhausted(target, reason);
+                if (startsWithExhaustion(matchedLine, exhausted)) {
+                    exhaustionSink.onExhausted(target, reason);
+                }
             }
             return true;
         }
@@ -492,8 +497,9 @@ public final class CompletionResolver implements TurnListener {
             if (rendezvous.resolveFailure(waiter, reason)) {
                 inFlight.remove(target, turn);
                 log.warn("failing send to {} via turn-stall fallback from the raw scrape: {}", target, reason);
-                // fleetd#201 Unit 1: only on the resolution that actually won the race.
-                backendErrorSink.onBackendError(target, backendError, reason);
+                if (startsWithBackendError(backendError, backendErrorPatternOrFallback(target))) {
+                    backendErrorSink.onBackendError(target, backendError, reason);
+                }
             }
             return true;
         }
@@ -540,10 +546,10 @@ public final class CompletionResolver implements TurnListener {
      * {@link #MIN_TURN_NANOS} — a crash signature (e.g. a backend HTTP 400 before the worker did
      * anything) that a bare {@code BUSY -> DONE} transition cannot be told apart from a genuinely
      * fast completion. Runs the same backend-error classification the normal and raw-scrape paths
-     * apply, against whatever is on screen right now: a match is a typed failure that notifies
-     * {@link #backendErrorSink} (only on the resolution that wins the race); a non-match stays the
-     * original generic too-fast failure, naming the member and both timings, with whatever the pane
-     * shows appended so the caller sees the cause, not just "it failed".
+     * apply, against whatever is on screen right now: a match together with the too-fast crash
+     * signature notifies {@link #backendErrorSink} (only on the resolution that wins the race). A
+     * non-match stays the original generic too-fast failure, naming the member and both timings,
+     * with whatever the pane shows appended so the caller sees the cause, not just "it failed".
      */
     private void failTooFast(String target, InFlight turn, CompletableFuture<Rendezvous.Resolution> waiter,
                               long elapsedNanos) {
@@ -611,6 +617,68 @@ public final class CompletionResolver implements TurnListener {
         return null;
     }
 
+    /**
+     * True when nothing before the match on this pane line ends a sentence — that is, the match is
+     * still inside the line's first sentence rather than inside prose a member wrote about it.
+     * Used to decide whether an exhaustion match may quarantine a credential (fleetd #348).
+     *
+     * <p><strong>Why this is looser than {@link #startsWithBackendError}.</strong> An
+     * {@code exhaustedPattern} is written per profile and may name only the decisive words of a
+     * provider message — {@code "usage limit has been reached"} without its leading {@code "The"}.
+     * A start-of-line check would then reject the genuine refusal. That is the false negative
+     * fleetd #348's invariant 1 calls the worse direction: an unrecorded exhaustion leaves the
+     * fleet spawning into a credential with no capacity, and a quarantine runs 1800s against the
+     * backend-error cooldown's fixed 60s.
+     *
+     * <p>This rule accepts a superset of what a start-of-line check accepts: if the match begins
+     * right after the chrome, there is nothing in front of it, so there is no sentence ending
+     * either. So moving to it cannot add a false negative.
+     *
+     * <p><strong>No chrome skipping here, deliberately.</strong> The first version of this method
+     * copied {@code startsWithBackendError}'s leading-chrome loop. Measured on merge: deleting that
+     * loop left all 1369 tests green, and it must — the scan only looks for {@code . ! ?}, and no
+     * terminal chrome character is one of those. A step that cannot change the result is worse than
+     * no step, because the next reader takes it as evidence that chrome was handled.
+     *
+     * <p>It stays a heuristic. Prose whose <em>first</em> sentence carries the pattern still
+     * notifies the sink, and a genuine refusal behind an earlier full stop (a hostname, a version
+     * number) still does not. Both are known and neither is fixed here.
+     */
+    private static boolean startsWithExhaustion(String line, Pattern pattern) {
+        var matcher = pattern.matcher(line);
+        if (!matcher.find()) {
+            return false;
+        }
+        for (int prefix = 0; prefix < matcher.start(); prefix++) {
+            if (".!?".indexOf(line.charAt(prefix)) >= 0) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * True when the error pattern begins the matched pane line, rather than appearing in prose.
+     *
+     * <p>Leading terminal chrome is skipped first — box-drawing characters, bullets, gutter bars and
+     * spaces. #339 introduced this check with a bare {@code lookingAt}, and that rejected a genuine
+     * error line rendered as {@code "| 503 Service Unavailable: ..."}: the send still failed, but the
+     * credential outage was never recorded. That is the false negative #339's own invariant 3 called
+     * worse than the false positive it set out to fix — measured with a throwaway probe on the
+     * raw-scrape path, which is exactly the path whose comment says to expect leading chrome.
+     *
+     * <p>Skipping only a leading run of non-letter, non-digit characters keeps the fix's intent. A
+     * member's prose ({@code "I checked the retry path. An API Error: makes it back off."}) still
+     * does not match, because there the pattern sits after words, not after chrome.
+     */
+    private static boolean startsWithBackendError(String line, Pattern pattern) {
+        int i = 0;
+        while (i < line.length() && !Character.isLetterOrDigit(line.charAt(i))) {
+            i++;
+        }
+        return pattern.matcher(line.substring(i)).lookingAt();
+    }
+
     /**
      * Coverage summary for the CB-578 stage A exhausted-pattern classification, logged at startup
      * the way {@link dev.ltms.fleet.health.FleetHealthMonitor#coverage} is — so an operator can

fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java

@@ -145,8 +145,58 @@ public final class Injector {
         return router != null ? router.agentsFor(target) : agents;
     }
 
+    /** The result of trying to remove an undelivered message from the injector. */
+    public enum Cancellation {
+        CANCELLED,
+        DELIVERED,
+        NOT_DELIVERED
+    }
+
+    /**
+     * An identity handle for one queued delivery. It is the only value accepted by
+     * {@link #cancel(Delivery)}, so a caller cannot cancel a different message with the same target
+     * or text.
+     */
+    public static final class Delivery {
+        private final Pending pending;
+
+        private Delivery(Pending pending) {
+            this.pending = pending;
+        }
+
+        public CompletableFuture<Void> completion() {
+            return pending.delivered;
+        }
+    }
+
     /** A pending message and the future that completes when it has been delivered. */
-    private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
+    private static final class Pending {
+        enum State { QUEUED, DELIVERED, NOT_DELIVERED, CANCELLED }
+
+        final String target;
+        final String text;
+        final TurnToken token;
+        final CompletableFuture<Void> delivered;
+        volatile State state = State.QUEUED; // written under the owning Target monitor
+
+        Pending(String target, String text, TurnToken token, CompletableFuture<Void> delivered) {
+            this.target = target;
+            this.text = text;
+            this.token = token;
+            this.delivered = delivered;
+        }
+
+        String text() {
+            return text;
+        }
+
+        TurnToken token() {
+            return token;
+        }
+
+        CompletableFuture<Void> delivered() {
+            return delivered;
+        }
     }
 
     /** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -177,15 +227,47 @@ public final class Injector {
      * <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
      * target" and "queue the message" and orphan it in a target it just removed.
      */
-    public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
+    public Delivery enqueue(String target, String text, TurnToken token) {
         CompletableFuture<Void> delivered = new CompletableFuture<>();
-        Pending p = new Pending(text, token, delivered);
+        Pending p = new Pending(target, text, token, delivered);
         targets.compute(target, (_, existing) -> {
             Target t = (existing != null) ? existing : new Target();
             t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
             return t;
         });
-        return delivered;
+        return new Delivery(p);
+    }
+
+    /**
+     * Cancel this exact queued delivery. The target monitor serializes this operation with
+     * {@link #onStatus}: if delivery wins that race, this returns {@link Cancellation#DELIVERED}
+     * rather than claiming the message remained queued.
+     */
+    public Cancellation cancel(Delivery delivery) {
+        Pending p = delivery.pending;
+        Target t = targets.get(p.target);
+        if (t == null) {
+            return cancellationOf(p);
+        }
+        synchronized (t) {
+            if (p.state != Pending.State.QUEUED || !t.queue.remove(p)) {
+                return cancellationOf(p);
+            }
+            p.state = Pending.State.CANCELLED;
+            if (isQuiescent(t)) {
+                targets.remove(p.target, t);
+            }
+            return Cancellation.CANCELLED;
+        }
+    }
+
+    private static Cancellation cancellationOf(Pending p) {
+        return p.state == Pending.State.DELIVERED ? Cancellation.DELIVERED : Cancellation.NOT_DELIVERED;
+    }
+
+    private static boolean isQuiescent(Target t) {
+        return t.queue.isEmpty() && !t.awaitingPickup && !t.awaitingCompletion
+                && !t.postTurnPending && !t.awaitingPostTurnPickup && !t.postTurnObserved;
     }
 
     /**
@@ -274,6 +356,7 @@ public final class Injector {
                             try {
                                 agentsFor(target).send(target, p.text());
                                 t.queue.poll();
+                                p.state = Pending.State.DELIVERED;
                                 t.awaitingPickup = true;
                                 t.awaitingCompletion = true;
                                 t.turnObserved = false;
@@ -283,6 +366,7 @@ public final class Injector {
                                 // Delivery failed at herdr; drop the poisoned message and surface it
                                 // rather than blocking the queue behind it.
                                 t.queue.poll();
+                                p.state = Pending.State.NOT_DELIVERED;
                                 sent = p;
                                 sendError = e;
                             }
@@ -293,6 +377,9 @@ public final class Injector {
                             // fail every queued message and release the target (CB-114) instead of
                             // polling it indefinitely with the caller's future never completing.
                             notReady = new ArrayList<>(t.queue);
+                            for (Pending pending : notReady) {
+                                pending.state = Pending.State.NOT_DELIVERED;
+                            }
                             log.warn("readiness grace for {} expired after {} polls ({}s): target never "
                                     + "became deliverable, so failing {} queued message(s) that never "
                                     + "reached its pane",
@@ -340,8 +427,7 @@ public final class Injector {
 
             // Reclaim the entry once the worker is fully quiescent (nothing queued, no pickup or
             // completion awaited), so the map cannot grow without bound across short-lived workers.
-            if (t.queue.isEmpty() && !t.awaitingPickup && !t.awaitingCompletion
-                    && !t.postTurnPending && !t.awaitingPostTurnPickup && !t.postTurnObserved) {
+            if (isQuiescent(t)) {
                 targets.remove(target, t);
             }
         }
@@ -432,6 +518,9 @@ public final class Injector {
         boolean hadDeliveredTurn;
         synchronized (t) {
             pending = new ArrayList<>(t.queue);
+            for (Pending p : pending) {
+                p.state = Pending.State.NOT_DELIVERED;
+            }
             t.queue.clear();
             hadDeliveredTurn = t.awaitingCompletion;
             t.awaitingCompletion = false;

fleetd/src/main/java/dev/ltms/fleet/member/HerdrPeerLauncher.java

@@ -937,14 +937,20 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      */
     @Override
     public void stop(String idOrPane) {
-        // Teardown knows only the pane, not which profile spawned it. Attempt tab cleanup when any
-        // profile uses tab placement (so the bridge may have created a dedicated peer tab); the
-        // single-occupant check below is what actually protects the user's shared tabs.
+        // Teardown knows only the pane, not which profile spawned it — and, when this call is
+        // routed here through CompositePeerLauncher's single-daemon stop() shortcut (fleetd #342),
+        // not even which adapter's config actually governed the spawn: the shortcut can hand the
+        // pane to a delegate that never spawned it, whose own profiles say nothing about how THIS
+        // pane was placed. So the decision to look for a tab to clean up is made from the pane's
+        // actual state, not from this delegate's static profile config: resolve the tab
+        // unconditionally and let {@link WorkspaceControl#locatePane} tolerate "not found" (it
+        // returns null rather than throwing); the single-occupant check below is what actually
+        // protects the user's shared tabs, exactly as it always has.
         String paneId = paneByAgentId.remove(idOrPane);
         if (paneId == null) {
             paneId = idOrPane; // raw-pane fallback (reap, gate timeout, pane-addressed callers)
         }
-        WorkspaceControl.PaneLocation loc = usesTabPlacement() ? spaces.locatePane(paneId) : null;
+        WorkspaceControl.PaneLocation loc = spaces.locatePane(paneId);
         try {
             agents.close(paneId);
         } catch (HerdrException e) {
@@ -985,11 +991,6 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
         }
     }
 
-    /** Whether any configured profile places peers in their own tab (so tabs may need cleanup). */
-    private boolean usesTabPlacement() {
-        return profiles.values().stream().anyMatch(FleetConfig.Profile::tabPlacement);
-    }
-
     /** True when a herdr error means the target is already gone (safe to treat as done). */
     private static boolean isAlreadyGone(HerdrException e) {
         return e.code() != null && e.code().endsWith("_not_found");
@@ -1665,30 +1666,59 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
 
     /**
      * Guards the {@code effectiveAllowed == null} branch of {@link #logCredentialGap} — the
-     * genuinely-unprotected report (deny-by-default, and the allow-list non-zsh fallback) — to one
-     * WARN per launcher instance, not one per spawn.
+     * genuinely-unprotected report (deny-by-default, and the allow-list non-zsh fallback) — AND
+     * the {@code effectiveAllowed != null} / {@code keptByDerivedList} branch, the allow-list case
+     * where a name is in the gap but the derived allow-list keeps it anyway. Both branches log the
+     * same severity (WARN) about the same fact — a name genuinely reaching a member pane
+     * unprotected — so they share this one guard, keyed per NAME rather than per launcher instance:
+     * each credential-shaped name that is ever reported unprotected gets exactly one WARN, however
+     * many spawns see it and whichever of the two branches first reports it.
+     *
+     * <p>fleetd #341: {@code memberCredentials} is a live, re-read-per-spawn supplier, so the
+     * policy — and so the gap's actual member names — can change between two spawns on the same
+     * launcher. Before this fix the guard was a single {@code AtomicBoolean} tripped by either
+     * branch: spawn 1 could warn about name A and trip the flag, and a later spawn's gap containing
+     * a different name B would never be reported, even though B is just as unprotected as A was.
+     * {@code AtomicBoolean} could not express "once per distinct name" at all — only "once, ever,
+     * for whichever name got there first" — so this is a {@code Set<String>} guard instead, the
+     * same shape {@link OpenCodeLauncher#modelCheckSkippedWarned} already uses for its own
+     * once-per-distinct-thing WARN. {@link #add}'s return value (true only the first time a name is
+     * added) is what turns "log the whole gap" into "log only the names never warned about before".
+     *
+     * <p>Bounded by construction: every name added here first passed {@link
+     * #CREDENTIAL_SHAPED_NAME}'s filter over {@link #hostEnvNames}, i.e. it is an actual
+     * environment variable name from the daemon's own process — a small, OS-bounded set (the host
+     * environment has, in practice, tens to a few hundred entries), not an attacker- or
+     * request-controlled input. So this set cannot grow past "however many distinct credential-
+     * shaped names this host's environment has ever held across this launcher's lifetime," which is
+     * effectively fixed for the life of one daemon process — no separate cap is needed.
      *
      * <p>CB-633 follow-up (#192): kept SEPARATE from {@link #allowListGapLogged} on purpose.
      * {@code memberCredentials} is a live, re-read-per-spawn supplier, so the policy can change
      * between two spawns on the same launcher. A single shared flag would let a harmless allow-list
      * INFO on spawn 1 permanently suppress the real deny-by-default WARN a later spawn deserves —
      * the report that matters most getting hidden by the report that doesn't. Two flags mean each
-     * report kind fires exactly once, independent of what the other kind already logged.
+     * report kind (WARN vs. INFO) fires independently of what the other kind already logged; within
+     * the WARN kind itself, the set above further separates by name, for the same reason.
      */
-    private final AtomicBoolean unprotectedGapLogged = new AtomicBoolean();
+    private final Set<String> unprotectedGapNamesWarned = ConcurrentHashMap.newKeySet();
 
     /**
      * Guards the {@code effectiveAllowed != null} branch of {@link #logCredentialGap} — the
      * allow-list-scrub-covered report — to one INFO per launcher instance. See {@link
-     * #unprotectedGapLogged}'s javadoc for why this is a separate flag rather than a shared one.
+     * #unprotectedGapNamesWarned}'s javadoc for why this is a separate flag rather than a shared
+     * one; unlike that guard it stays a per-instance {@code AtomicBoolean}, not a per-name set —
+     * fleetd #341 fixed the WARN-vs-WARN suppression, not this INFO's own one-shot shape, which was
+     * not reported as broken and is out of that ticket's scope.
      */
     private final AtomicBoolean allowListGapLogged = new AtomicBoolean();
 
     /**
      * fleetd #185 stage 2: guards {@link #warnUnknownMemberEnvironment} to one WARN per launcher
-     * instance, not one per spawn — the same one-per-instance shape as {@link #unprotectedGapLogged}
-     * and {@link #allowListGapLogged}, kept as its own flag for the same reason those two are split:
-     * this mode is orthogonal to which of the other two branches would otherwise have fired.
+     * instance, not one per spawn — the same one-shot shape {@link #unprotectedGapNamesWarned} and
+     * {@link #allowListGapLogged} guard their own branches with, kept as its own flag for the same
+     * reason those two are split: this mode is orthogonal to which of the other two branches would
+     * otherwise have fired.
      */
     private final AtomicBoolean unknownMemberEnvironmentWarned = new AtomicBoolean();
 
@@ -1816,14 +1846,21 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
         List<String> blankedByScrub = gap.stream()
                 .filter(name -> !MemberEnvAllowList.keeps(effectiveAllowed, name))
                 .toList();
-        if (!keptByDerivedList.isEmpty() && unprotectedGapLogged.compareAndSet(false, true)) {
+        // fleetd #341: filter to names this guard has never warned about before — not just
+        // "isEmpty" on the whole branch — so a name this spawn's gap shares with an EARLIER
+        // spawn's (already-warned) gap does not re-print, while a name unique to THIS gap still
+        // does, whichever of the two WARN branches reported it first.
+        List<String> newlyUnprotected = keptByDerivedList.stream()
+                .filter(unprotectedGapNamesWarned::add)
+                .toList();
+        if (!newlyUnprotected.isEmpty()) {
             log.warn("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
                             + "known: nor allow: — the derived allow-list keeps them anyway (a profile's "
                             + "gitTokenEnv/gitHostEnv/tokenEnv/env: names one, or this spawn injects it), "
                             + "so every member pane inherits them UNBLOCKED — {}. Add each to "
                             + "memberCredentials.known (or .allow if a member legitimately needs it), or "
                             + "remove it from whatever profile setting derives it in.",
-                    keptByDerivedList.size(), keptByDerivedList);
+                    newlyUnprotected.size(), newlyUnprotected);
         }
         if (!blankedByScrub.isEmpty() && allowListGapLogged.compareAndSet(false, true)) {
             log.info("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
@@ -1836,12 +1873,18 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
 
     /** The deny-by-default (and allow-list non-zsh fallback) WARN — unchanged byte-for-byte by #192. */
     private void warnGapUnprotected(List<String> gap) {
-        if (unprotectedGapLogged.compareAndSet(false, true)) {
+        // fleetd #341: same "only the names never warned before" filter as the sibling branch in
+        // logCredentialGap above — see unprotectedGapNamesWarned's javadoc. Both branches share
+        // this one guard because both report the exact same fact (a name reaching a member pane
+        // unprotected) at the exact same severity; keying it by name is what lets a later spawn's
+        // DIFFERENT name still get its own WARN after an earlier spawn's already fired.
+        List<String> newlyUnprotected = gap.stream().filter(unprotectedGapNamesWarned::add).toList();
+        if (!newlyUnprotected.isEmpty()) {
             log.warn("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
                             + "known: nor allow: — every member pane inherits them UNBLOCKED — {}. "
                             + "Add each to memberCredentials.known (blocked by default) or .allow "
                             + "(if a member legitimately needs it).",
-                    gap.size(), gap);
+                    newlyUnprotected.size(), newlyUnprotected);
         }
     }
 

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

@@ -8,6 +8,7 @@ import com.rabbitmq.client.DeliverCallback;
 import com.rabbitmq.client.Recoverable;
 import com.rabbitmq.client.RecoveryListener;
 import com.rabbitmq.client.Return;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -153,17 +154,22 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     /** 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("fleetd-reply-inbox"), prefetch);
+            return new AmqpReplyInbox(connectionFactory(uri).newConnection(AmqpConnectionFailureLogger.REPLY_INBOX), prefetch);
         } catch (Exception e) {
             throw new IllegalStateException("cannot connect to AMQP broker at " + uri, e);
         }
     }
 
+    static ConnectionFactory connectionFactory(String uri) throws Exception {
+        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);
+        factory.setExceptionHandler(new AmqpConnectionFailureLogger(AmqpConnectionFailureLogger.REPLY_INBOX, log));
+        return factory;
+    }
+
     /** Wrap an already-open connection with {@link #DEFAULT_PREFETCH} (injection seam for the contract test). */
     AmqpReplyInbox(Connection connection) {
         this(connection, DEFAULT_PREFETCH);
@@ -571,3 +577,44 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
         }
     }
 }
+
+/**
+ * Keeps RabbitMQ's forgiving exception behaviour while adding the connection identity that its
+ * default logger drops. Package-private so both AMQP connections use the same two names.
+ */
+final class AmqpConnectionFailureLogger extends DefaultExceptionHandler {
+
+    static final String REPLY_INBOX = "fleetd-reply-inbox";
+    static final String LEAD_MAILBOX = "fleetd-lead-mailbox";
+
+    private final String connectionName;
+    private final Logger logger;
+
+    AmqpConnectionFailureLogger(String connectionName, Logger logger) {
+        this.connectionName = connectionName;
+        this.logger = logger;
+    }
+
+    String connectionName() {
+        return connectionName;
+    }
+
+    @Override
+    protected void log(String message, Throwable cause) {
+        if (isSocketClosedOrConnectionReset(cause)) {
+            logger.warn("AMQP connection {}: {} (Exception message: {})", connectionName, message, cause.getMessage());
+        } else {
+            logger.error("AMQP connection {}: {}", connectionName, message, cause);
+        }
+    }
+
+    private static boolean isSocketClosedOrConnectionReset(Throwable cause) {
+        // Deliberate copy of ForgivingExceptionHandler's private static helper; check it on amqp-client upgrades.
+        if (!(cause instanceof IOException)) {
+            return false;
+        }
+        return "Connection reset".equals(cause.getMessage())
+                || "Socket closed".equals(cause.getMessage())
+                || "Connection reset by peer".equals(cause.getMessage());
+    }
+}

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

@@ -122,17 +122,22 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
     /** As {@link #open(String, String)}, with an explicit consumer prefetch. */
     public static LeadMailbox open(String uri, String selfCoordId, int prefetch) {
         try {
-            ConnectionFactory factory = new ConnectionFactory();
-            factory.setUri(uri);
-            // Self-heal transient blips; topology recovery re-declares the queue and re-attaches the consumer.
-            factory.setAutomaticRecoveryEnabled(true);
-            factory.setTopologyRecoveryEnabled(true);
-            return new LeadMailbox(factory.newConnection("fleetd-lead-mailbox"), selfCoordId, prefetch);
+            return new LeadMailbox(connectionFactory(uri).newConnection(AmqpConnectionFailureLogger.LEAD_MAILBOX), selfCoordId, prefetch);
         } catch (Exception e) {
             throw new IllegalStateException("cannot connect to AMQP coordination broker at " + uri, e);
         }
     }
 
+    static ConnectionFactory connectionFactory(String uri) throws Exception {
+        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);
+        factory.setExceptionHandler(new AmqpConnectionFailureLogger(AmqpConnectionFailureLogger.LEAD_MAILBOX, log));
+        return factory;
+    }
+
     /** Wrap an already-open connection with {@link #DEFAULT_PREFETCH} (injection seam for tests). */
     LeadMailbox(Connection connection, String selfCoordId) {
         this(connection, selfCoordId, DEFAULT_PREFETCH);

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

@@ -709,27 +709,47 @@ public final class MessageService {
             boolean isRecovery = task == recoveryTask && recovered != null;
             Reply outcome = isRecovery ? recovered : new Reply(Outcome.WORKER_FAILED, reason);
             String turnId = task.turnId;
-            if (task.future.complete(outcome)) {
-                if (outcome.outcome() == Outcome.WORKER_FAILED) {
-                    asyncFailed = true;
+            // fleetd #335: completing THIS task's future must not depend on any other task's
+            // cleanup succeeding — every task in `matching` is owed its own outcome regardless of
+            // what happens below, so decide and record that before doing anything that can throw.
+            boolean completedHere = task.future.complete(outcome);
+            if (completedHere && outcome.outcome() == Outcome.WORKER_FAILED) {
+                asyncFailed = true;
+            }
+            try {
+                if (abandonCleanupHookForTest != null) {
+                    // Test-only (fleetd #335 site 1): see the field's own javadoc.
+                    abandonCleanupHookForTest.run();
                 }
-                if (turnId != null) {
-                    // #275: whether this task was swept out of ASKING or was already answered and
-                    // only waiting on its resumed turn's real reply (#137), nothing will ever
-                    // complete this turnId now — drop it from this class's own bookkeeping AND the
-                    // reverse-rendezvous itself, so hasAsyncQuestion(target) stops reporting a turn
-                    // that is actually done, and a late answer() sees it as lapsed rather than
-                    // resolving a question nothing is listening for any more.
-                    asyncTasksByTurn.remove(turnId, task);
-                    rendezvous.closeAsk(turnId);
+                if (completedHere) {
+                    if (turnId != null) {
+                        // #275: whether this task was swept out of ASKING or was already answered and
+                        // only waiting on its resumed turn's real reply (#137), nothing will ever
+                        // complete this turnId now — drop it from this class's own bookkeeping AND the
+                        // reverse-rendezvous itself, so hasAsyncQuestion(target) stops reporting a turn
+                        // that is actually done, and a late answer() sees it as lapsed rather than
+                        // resolving a question nothing is listening for any more.
+                        asyncTasksByTurn.remove(turnId, task);
+                        rendezvous.closeAsk(turnId);
+                    }
+                } else if (isRecovery) {
+                    // The recovered reply was already drained out of the inbox, but this task resolved
+                    // through another path (e.g. a concurrent reply() or a second abandon() racing this
+                    // one) between us choosing it and completing it here. Put the reply back rather than
+                    // lose it silently — it may still belong to some other still-open task, or the next
+                    // caller that drains this target's inbox.
+                    inbox.publish(target, UUID.randomUUID().toString(), recovered.text());
                 }
-            } else if (isRecovery) {
-                // The recovered reply was already drained out of the inbox, but this task resolved
-                // through another path (e.g. a concurrent reply() or a second abandon() racing this
-                // one) between us choosing it and completing it here. Put the reply back rather than
-                // lose it silently — it may still belong to some other still-open task, or the next
-                // caller that drains this target's inbox.
-                inbox.publish(target, UUID.randomUUID().toString(), recovered.text());
+            } catch (RuntimeException e) {
+                // fleetd #335: inbox.publish reaches a broker (AmqpReplyInbox throws
+                // IllegalStateException on an unroutable/unconfirmed/interrupted publish) and this
+                // loop has no other teardown path — a caller on the release path, or the health
+                // monitor's GONE/NEVER_READY sweep. Losing this exception uncaught would abort the
+                // loop and leave every task still to come in `matching` PENDING forever (fleetd
+                // #335 site 1). completedHere is already recorded above, so only this task's
+                // best-effort bookkeeping is lost — log it and let the loop reach the rest.
+                log.error("abandon: per-task cleanup failed for ticket {} (target {}, turnId {})",
+                        task.ticket, target, turnId, e);
             }
         }
         if (failed) {
@@ -863,16 +883,26 @@ public final class MessageService {
                 if (onAccepted != null) {
                     onAccepted.run();
                 }
-                CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
+                Injector.Delivery delivery = injector.enqueue(target, content, token);
                 try {
                     Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
                     return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
                 } catch (TimeoutException e) {
-                    boolean wasDelivered = delivered.isDone() && !delivered.isCompletedExceptionally();
+                    boolean wasDelivered = delivery.completion().isDone()
+                            && !delivery.completion().isCompletedExceptionally();
+                    if (!wasDelivered) {
+                        if (timeoutCancellationRaceHookForTest != null) {
+                            // Test-only (fleetd #345): see the field's own javadoc.
+                            timeoutCancellationRaceHookForTest.run();
+                        }
+                        // The target monitor makes cancellation atomic with onStatus picking this
+                        // Pending up. If pickup won, report TIMED_OUT_WORKING because the text landed.
+                        wasDelivered = injector.cancel(delivery) == Injector.Cancellation.DELIVERED;
+                    }
                     log.debug("send to {} timed out (delivered={})", target, wasDelivered);
                     if (!wasDelivered) {
-                        // CB-640: still sitting in the injector's queue, waiting for the member to
-                        // go idle — record the fact for fleet health (see queuedDeliveries).
+                        // CB-640: record that delivery did not happen for fleet health (see
+                        // queuedDeliveries). The exact Pending was cancelled, so it cannot arrive later.
                         queuedDeliveries.put(target, Boolean.TRUE);
                     }
                     return recorded(new Reply(
@@ -935,13 +965,41 @@ public final class MessageService {
             return new AskResult(AskOutcome.ANSWERED, answer);
         } catch (TimeoutException e) {
             log.debug("fleet_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
-            // fleetd #307: mark the task BEFORE clearAsyncQuestion(forgetTurn=true) below drops it out of
-            // asyncTasksByTurn and nulls its turnId — that forgetting is deliberate and stays (it is
-            // what keeps the target from staying BUSY forever), but it would otherwise also erase
-            // askAnsweredAsyncTasks' only signal that the worker's eventual real fleet_reply still
-            // belongs to this task, stranding it in the inbox with a false "never replied" verdict.
-            markAskTimedOut(ticket.turnId());
-            clearAsyncQuestion(ticket.turnId(), true);
+            // Only the fresh owner tears down the shared turn (mirrors the finally block below).
+            // A duplicate's own timeoutMillis says nothing about whether the SHARED ask is actually
+            // done — it must leave the close/forget bookkeeping to the fresh owner, exactly as it
+            // already leaves closeAsk to it.
+            if (ticket.fresh()) {
+                // fleetd #334: close the ask turn BEFORE forgetting this task's turnId mapping below.
+                // Before this fix the order was reversed — the mapping was forgotten here first, and
+                // rendezvous.closeAsk only ran afterward, in the shared finally. A primary's answer()
+                // call racing this exact timeout could then find rendezvous.askSession(turnId) still
+                // non-null (the ask still "answerable") after the Task mapping was already gone:
+                // answer()'s own asyncTasksByTurn lookup returned null, its task != null guard skipped
+                // the completion, and the async ticket sat at PENDING forever even though answer()
+                // itself reported the worker's real reply. Closing here first removes that window:
+                // any answer() call that still observes askSession(turnId) != null is necessarily
+                // racing a point BEFORE the forgetting below runs (both happen on this one thread, in
+                // this order, with nothing that yields in between), so the Task mapping is still there
+                // for it to find; any call that observes askSession(turnId) == null now correctly
+                // bails out STALE_TURN (see answer()'s own top check) before ever reaching
+                // asyncTasksByTurn. rendezvous.closeAsk is idempotent — a no-op once the turn is
+                // already removed, see its own javadoc — so the shared finally below re-running it
+                // for this same fresh call is harmless.
+                rendezvous.closeAsk(ticket.turnId());
+                if (askTimeoutRaceHookForTest != null) {
+                    // Test-only (fleetd #334): see the field's own javadoc.
+                    askTimeoutRaceHookForTest.run();
+                }
+                // fleetd #307: mark the task BEFORE clearAsyncQuestion(forgetTurn=true) below drops it
+                // out of asyncTasksByTurn and nulls its turnId — that forgetting is deliberate and
+                // stays (it is what keeps the target from staying BUSY forever), but it would
+                // otherwise also erase askAnsweredAsyncTasks' only signal that the worker's eventual
+                // real fleet_reply still belongs to this task, stranding it in the inbox with a false
+                // "never replied" verdict.
+                markAskTimedOut(ticket.turnId());
+                clearAsyncQuestion(ticket.turnId(), true);
+            }
             return new AskResult(AskOutcome.TIMED_OUT, null);
         } catch (ExecutionException e) {
             Throwable cause = e.getCause();
@@ -1040,19 +1098,28 @@ public final class MessageService {
                 // the chained ask deliberately left open.
                 //
                 // A null task is NOT only "this was never an async ticket". That reading was in this
-                // comment when #329 merged and it is wrong. A genuine async ticket also lands here
-                // with task == null, because ask()'s timeout path runs clearAsyncQuestion(turnId,
-                // true) — which drops the asyncTasksByTurn entry — in its catch block, while
-                // rendezvous.closeAsk(turnId) runs later, in its finally. Between those two the ask
-                // is still answerable but the map entry is already gone, so the lookup at :991
-                // returns null and this ticket is never completed. Measured on 2026-09-04: a probe
-                // firing only that first half before answer() runs printed
-                // "answer=REPLIED phase=PENDING reply=null" — the same stranded ticket #329 set out
-                // to fix, one step earlier in the same race. The probe used forgetTurnForTest, which
-                // omits ask()'s markAskTimedOut; that cannot change the outcome, because askTimedOut
-                // is read only by askAnsweredAsyncTasks, and reply() never reaches it while this
-                // method's own waiter is live. So #329 narrows this window rather than closing it.
-                // Open as fleetd #334 — do not read this guard as complete.
+                // comment when #329 merged and it is wrong; it is still not the whole story after
+                // #334. A genuine async ticket can still land here with task == null — a blocking
+                // (wait:true) send's fleet_ask never has a Task at all, so that case is expected and
+                // fine. What #334 fixed was a SECOND, unintended way to get here with task == null:
+                // ask()'s timeout path used to run clearAsyncQuestion(turnId, true) — which drops the
+                // asyncTasksByTurn entry — in its catch block, while rendezvous.closeAsk(turnId) ran
+                // later, in its finally. Between those two the ask was still answerable but the map
+                // entry was already gone, so the lookup at :1053 returned null and this ticket was
+                // never completed. Measured on 2026-09-04: a probe firing only that first half before
+                // answer() ran printed "answer=REPLIED phase=PENDING reply=null" — the same stranded
+                // ticket #329 set out to fix, one step earlier in the same race; the probe used
+                // forgetTurnForTest, which omits ask()'s markAskTimedOut, and that omission does not
+                // change the outcome, because askTimedOut is read only by askAnsweredAsyncTasks, and
+                // reply() never reaches it while this method's own waiter is live. #334's fix
+                // reorders ask()'s timeout catch to run closeAsk before the forgetting (see the
+                // fresh-owner block there), which removes this path entirely rather than narrowing
+                // it further: once closeAsk has run, rendezvous.askSession(turnId) is null and
+                // answer() returns STALE_TURN from its own top check, before it ever reaches this
+                // lookup — see aLateAnswerDuringAskTimeoutTeardownStillCompletesTheAsyncTicket in
+                // MessageServiceTest, which pins the exact window with askTimeoutRaceHookForTest.
+                // So by the time this line runs, task == null means only the ordinary blocking-send
+                // case (or #329's own already-fixed race elsewhere) — not this one.
                 if (result.outcome() != Outcome.QUESTION && task != null) {
                     finishAsyncTask(task, result);
                 }
@@ -1112,7 +1179,18 @@ public final class MessageService {
             // class javadoc on sendAsync/CB-107.
             task.future.whenComplete((reply, ex) -> {
                 boolean failed = ex != null || reply == null || !reply.completed();
-                pushLoop.onTicketTerminal(ticket, target, failed);
+                try {
+                    pushLoop.onTicketTerminal(ticket, target, failed);
+                } catch (Throwable t) {
+                    // fleetd #335 (site 2): this stage's own CompletableFuture is discarded, so an
+                    // uncaught throw here (e.g. a RejectedExecutionException from
+                    // ReplyPushLoop's scheduler, already shut down while this in-flight send's
+                    // whenComplete fires during the daemon's own shutdown sequence — messages.close()
+                    // only stops accepting NEW async work, it does not cancel a delivery already
+                    // running) vanishes with no log line and no metric, and the push loop never learns
+                    // the ticket went terminal — the exact thing this hook exists to tell it.
+                    log.error("push loop failed to learn ticket {} (target {}) went terminal", ticket, target, t);
+                }
             });
         }
         asyncExecutor.submit(() -> {
@@ -1364,6 +1442,24 @@ public final class MessageService {
         this.afterFinishAsyncTaskCompleteHookForTest = hook;
     }
 
+    /**
+     * Null in production; test seam for fleetd #345 — invoked in {@link #send}'s timeout path after
+     * {@link Injector.Delivery#completion()} reports incomplete and before {@link Injector#cancel}
+     * takes the target monitor. A test installs this to make {@code onStatus} pick the exact queued
+     * delivery up in that window, so {@code cancel} returns {@link Injector.Cancellation#DELIVERED}.
+     * This deterministically covers the caller's need to use that result rather than relying on a
+     * timing-sensitive real race.
+     */
+    private volatile Runnable timeoutCancellationRaceHookForTest;
+
+    /**
+     * Test-only (fleetd #345): install {@link #timeoutCancellationRaceHookForTest}. Package-private
+     * so the test, in the same package, can reach it without widening any production API.
+     */
+    void setTimeoutCancellationRaceHookForTest(Runnable hook) {
+        this.timeoutCancellationRaceHookForTest = hook;
+    }
+
     /**
      * Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
      * the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
@@ -1393,6 +1489,56 @@ public final class MessageService {
         clearAsyncQuestion(turnId, true);
     }
 
+    /**
+     * Null in production; test seam for fleetd #335 (site 1) — invoked from {@link #abandon(String,
+     * String, boolean)}'s per-task loop, once per task, right before that task's own cleanup
+     * (turnId bookkeeping, or the stranded-reply put-back) runs. A test installs this to inject a
+     * throw at that exact point deterministically.
+     *
+     * <p>The one production call there that can really throw is {@code inbox.publish} in the
+     * put-back branch — {@link AmqpReplyInbox#publish} reaches a broker and throws {@link
+     * IllegalStateException} on an unroutable, unconfirmed, or interrupted publish — but reaching
+     * that branch requires a second completion of the very task {@code abandon} is about to
+     * complete to win the race first (see the branch's own comment), and the #137 follow-up
+     * investigation above already found the combination this needs (a stranded reply coinciding
+     * with an open matching task) unreachable through the public API, not merely hard to time.
+     * This hook reproduces the resulting shape — a per-task cleanup throw — directly, the same
+     * technique {@link #finishAsyncTaskRaceHook} and {@link
+     * #afterFinishAsyncTaskCompleteHookForTest} already use for their own hard-to-time races.
+     */
+    private volatile Runnable abandonCleanupHookForTest;
+
+    /**
+     * Test-only (fleetd #335, site 1): install {@link #abandonCleanupHookForTest}. Package-private
+     * so the test, in the same package, can reach it without widening any production API.
+     */
+    void setAbandonCleanupHookForTest(Runnable hook) {
+        this.abandonCleanupHookForTest = hook;
+    }
+
+    /**
+     * Null in production; test seam for fleetd #334 — invoked from {@link #ask}'s {@code
+     * TimeoutException} catch, only for the fresh owner, right after {@code rendezvous.closeAsk}
+     * has run and before {@link #markAskTimedOut} / {@link #clearAsyncQuestion} forget this task's
+     * turnId mapping. A test installs this to call {@link #answer} for the very same {@code turnId}
+     * synchronously from inside that exact window, deterministically reproducing the race a real
+     * concurrent {@code answer()} call could otherwise only win by timing luck: with the ask already
+     * closed, that call must see {@code rendezvous.askSession(turnId) == null} and return {@link
+     * Outcome#STALE_TURN} immediately, never reaching {@code asyncTasksByTurn} at all — proving the
+     * window fleetd #334 describes (mapping forgotten while the ask was still "answerable") is
+     * closed, rather than merely narrowed the way fleetd #329 narrowed the sibling race in {@link
+     * #finishAsyncTask}.
+     */
+    private volatile Runnable askTimeoutRaceHookForTest;
+
+    /**
+     * Test-only (fleetd #334): install {@link #askTimeoutRaceHookForTest}. Package-private so the
+     * test, in the same package, can reach it without widening any production API.
+     */
+    void setAskTimeoutRaceHookForTest(Runnable hook) {
+        this.askTimeoutRaceHookForTest = hook;
+    }
+
     /** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
     private boolean hasAsyncQuestion(String target) {
         return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));

fleetd/src/test/java/dev/ltms/fleet/config/FleetConfigWithDefaultsPreservesEveryComponentTest.java

@@ -0,0 +1,192 @@
+package dev.ltms.fleet.config;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.RecordComponent;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Guards against a "defect factory" built into this file's own established pattern, found live
+ * while building the (parked) idle-sleep-guard PR: every time a component is added to
+ * {@link FleetConfig}, the record grows by one arg AND a new back-compat constructor is added at
+ * the OLD arity, so existing callers keep compiling. That is correct and required — see the
+ * constructor ladder just below the record header. But {@link #withDefaults()}'s own {@code return
+ * new FleetConfig(...)} call sits in this same file, written at a literal argument count. The very
+ * next time a component is added, the freshly-added back-compat constructor at the OLD arity
+ * silently captures that stale call, because it is now a legal overload at that arg count too. It
+ * compiles. Every other test passes, because nothing else exercises the new field. The new
+ * component is defaulted away — {@code null}, or whatever that back-compat overload defaults it to
+ * — on every {@link FleetConfig#load}. Measured, not theoretical: this exact sequence happened
+ * live when the {@code idleSleepGuard} component was added on a sibling branch; it was caught only
+ * because that branch's own new tests happened to assert on the new field's value.
+ *
+ * <p>This test proves the opposite property, and does it in a way that survives the next field
+ * being added without being rewritten: reflectively enumerate {@link FleetConfig}'s own record
+ * components (never a hardcoded count — the arity is exactly what changes over time), build one
+ * config through the true canonical constructor with a real, distinctive, non-null value in EVERY
+ * component (reusing the exact reflective-construction pattern
+ * {@link ConfigRefTopLevelReportingCoverageTest} already established for this file:
+ * {@code getDeclaredConstructor(exact record-component types)}, which resolves the canonical
+ * constructor by its true shape, not by binding to whichever overload happens to match arg count —
+ * the same way Jackson resolves it), call the real {@link FleetConfig#withDefaults()}, and assert
+ * every one of those values survives unchanged.
+ *
+ * <p>Why this is a valid check for every component, not just some: {@link #withDefaults()}'s own
+ * comments document that it only ever REPLACES a component when the incoming value is {@code null}
+ * (or blank, for {@code placement}) — {@code broker}/{@code primary}/{@code leadHeartbeat}/
+ * {@code configReload}/{@code coordinator}/{@code worktreeGroup}/{@code memberLoginShell} are left
+ * as-is unconditionally, and {@code bind}/{@code guard}/{@code lifecycle}/{@code auth}/
+ * {@code fleet}/{@code quarantineCooldownSeconds}/{@code memberCredentials}/{@code placement} are
+ * replaced only on null/blank input. A value that is never null or blank going in must therefore
+ * never change coming out, for every current component. No exclusion is needed today.
+ *
+ * <p>{@link #EXCLUDED_FROM_SURVIVAL_CHECK} exists anyway, kept deliberately empty and size-pinned
+ * by {@link #exclusionListSizeIsPinned()}: a future component that {@code withDefaults()} is
+ * <em>documented</em> to transform unconditionally (unlike every field today) would legitimately
+ * need one. Pinning the size at 0 means growing that set to make a failure go away is itself a
+ * visible diff to this test, not a silent one — a checker that can be silenced by adding to its
+ * own escape hatch is not a checker.
+ */
+class FleetConfigWithDefaultsPreservesEveryComponentTest {
+
+    private static final RecordComponent[] COMPONENTS = FleetConfig.class.getRecordComponents();
+
+    /** See the class javadoc — deliberately empty today; grow it only with a matching justification. */
+    private static final Set<String> EXCLUDED_FROM_SURVIVAL_CHECK = Set.of();
+
+    /** One real, distinctive, non-null (non-blank where blankness would mean "unset") value per component. */
+    private static Map<String, Object> baseValues() {
+        Map<String, Object> v = new LinkedHashMap<>();
+        v.put("bind", new FleetConfig.Bind("127.0.0.1", 8765));
+        v.put("herdrSocket", "~/.config/herdr/guard.sock");
+        v.put("memberHerdrSocket", "~/.config/herdr/member-guard.sock");
+        v.put("profiles", Map.of("sonnet", minimalProfile("sonnet")));
+        v.put("guard", new FleetConfig.Guard(List.of("host-guard")));
+        v.put("worktreeRoot", "/wt/guard");
+        v.put("lifecycle", new FleetConfig.Lifecycle(300, 5, 30, true));
+        v.put("spawnReadyTimeoutMs", 12_345);
+        v.put("spawnReadyPollMs", 234);
+        v.put("broker", new FleetConfig.Broker("amqp://guard", null, 7));
+        v.put("primary", new FleetConfig.Primary("term-guard", 4, 4000));
+        v.put("fleet", new FleetConfig.Fleet(
+                Map.of("opus", new FleetConfig.Leader("sonnet", "lead: opus-guard", 1, null, 10,
+                        "claude", null, null, null)),
+                Map.of(), Map.of(), Map.of(), Map.of(), "{role}: {profile} #{n}"));
+        v.put("leadHeartbeat", new FleetConfig.LeadHeartbeat(301, 61_000L, 4));
+        v.put("health", new FleetConfig.Health(true, 31, 601, 61, null));
+        v.put("placement", "round-robin");
+        v.put("auth", new FleetConfig.Auth("loopback-trust", null));
+        v.put("configReload", new FleetConfig.ConfigReload(true, 11));
+        v.put("quarantineCooldownSeconds", 1801);
+        v.put("memberCredentials", new FleetConfig.MemberCredentials(
+                FleetConfig.MemberCredentials.POLICY_DENY_BY_DEFAULT,
+                List.of("git"), List.of("git", "ssh"), null));
+        v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-guard", null, "self-guard", 3));
+        v.put("worktreeGroup", "group-guard");
+        v.put("memberLoginShell", "/bin/zsh");
+        assertNamesMatchComponents(v);
+        return v;
+    }
+
+    /** A minimal, otherwise-null {@link FleetConfig.Profile} — just enough to name one in a map. */
+    private static FleetConfig.Profile minimalProfile(String name) {
+        return new FleetConfig.Profile(name, null, null, null, null, null, null, null, null, null,
+                null, null, null, null, null, null, null, null, null, null, null, null, null, null,
+                null, null);
+    }
+
+    /**
+     * Guards {@link #baseValues()} itself against drifting from the record's real shape — the same
+     * assurance {@link ConfigRefTopLevelReportingCoverageTest} already relies on. This is what makes
+     * "no hardcoded arity" true in practice: forgetting to add a new component here fails this
+     * assertion by name, rather than silently checking one component fewer than the record has.
+     */
+    private static void assertNamesMatchComponents(Map<String, Object> values) {
+        Set<String> names = new TreeSet<>();
+        for (RecordComponent rc : COMPONENTS) {
+            names.add(rc.getName());
+        }
+        assertEquals(names, new TreeSet<>(values.keySet()),
+                "this test's value map has drifted from FleetConfig's actual top-level components — "
+                        + "update baseValues() alongside the record");
+    }
+
+    /**
+     * Builds a {@link FleetConfig} through the TRUE canonical constructor — resolved by the record's
+     * own component types, not by argument count — so this never accidentally exercises a
+     * back-compat overload the way a literal {@code new FleetConfig(...)} call risks doing.
+     */
+    private static FleetConfig configOf(Map<String, Object> values) throws ReflectiveOperationException {
+        Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
+        Object[] args = Arrays.stream(COMPONENTS).map(rc -> values.get(rc.getName())).toArray();
+        Constructor<FleetConfig> ctor = FleetConfig.class.getDeclaredConstructor(types);
+        return ctor.newInstance(args);
+    }
+
+    @Test
+    void exclusionListSizeIsPinned() {
+        assertEquals(0, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
+                "EXCLUDED_FROM_SURVIVAL_CHECK grew from 0 — every entry needs a justification in "
+                        + "this test class's javadoc AND this assertion re-pinned to the new size; a "
+                        + "growing exclusion list that silences failures on its own is not a guard");
+    }
+
+    /**
+     * The mutation this is built to catch: make {@code withDefaults()}'s final constructor call
+     * literal at some arg count, add one more component to the record with a new back-compat
+     * constructor at the old arity, and the stale call silently rebinds. Every component here is
+     * real and non-null (non-blank for the one String — {@code placement} — where blank has
+     * meaning), so none of it should be replaced by {@code withDefaults()}; any component that
+     * comes back different was silently dropped.
+     */
+    @Test
+    void everyComponentGivenARealValueSurvivesWithDefaults() throws ReflectiveOperationException {
+        Map<String, Object> base = baseValues();
+        FleetConfig config = configOf(base);
+        FleetConfig defaulted = config.withDefaults();
+
+        List<String> dropped = new ArrayList<>();
+        int checked = 0;
+        for (RecordComponent rc : COMPONENTS) {
+            String name = rc.getName();
+            if (EXCLUDED_FROM_SURVIVAL_CHECK.contains(name)) {
+                continue;
+            }
+            checked++;
+            Object expected = base.get(name);
+            Object actual;
+            try {
+                actual = rc.getAccessor().invoke(defaulted);
+            } catch (ReflectiveOperationException e) {
+                throw new RuntimeException("failed to read FleetConfig." + name + "()", e);
+            }
+            if (!Objects.equals(expected, actual)) {
+                dropped.add(String.format(Locale.ROOT,
+                        "%s: withDefaults() was given a real, non-null value (%s) for '%s' but "
+                                + "returned %s — a component silently dropped by withDefaults(), the "
+                                + "shape of the defect this test exists to catch (its final "
+                                + "\"return new FleetConfig(...)\" call binding to a back-compat "
+                                + "constructor instead of the true canonical one)",
+                        name, expected, name, actual));
+            }
+        }
+
+        System.out.printf(Locale.ROOT,
+                "FleetConfig.withDefaults() component-survival coverage — %d components, %d checked, "
+                        + "%d excluded, %d survived%n",
+                COMPONENTS.length, checked, EXCLUDED_FROM_SURVIVAL_CHECK.size(), checked - dropped.size());
+        assertEquals(List.of(), dropped,
+                "withDefaults() silently dropped these real, given components: " + dropped);
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java

@@ -484,6 +484,27 @@ class CompletionResolverTest {
 
     // --- CB-578 stage A: backend-exhausted classification ---------------------------------
 
+    @Test
+    void aNormalMemberReportMentioningTheExhaustionPatternDoesNotNotifyTheSink() {
+        String block = "⏺ I reviewed capacity handling. The usage limit has been reached means no more work can start.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
+                "a matching report still fails the send as exhausted");
+        assertTrue(waiter.getNow(null).text().contains("I reviewed capacity handling."),
+                "the exhausted result keeps the whole matched pane line");
+        assertTrue(notified.isEmpty(),
+                "a normal report mentioning an exhaustion pattern must not quarantine a credential");
+    }
+
     @Test
     void classifiesAMatchingScrapeAsBackendExhaustedInsteadOfACompletedReply() {
         String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
@@ -534,6 +555,53 @@ class CompletionResolverTest {
                 "the sink is told the matched reason: " + notified.get(0));
     }
 
+    @Test
+    void aRealExhaustionBehindTerminalChromeStillNotifiesTheSink() {
+        String block = "⏺ │ The usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
+                "a real exhaustion must still fail the send as exhausted");
+        assertEquals(1, notified.size(),
+                "a real exhaustion behind terminal chrome must reach the sink");
+    }
+
+    /**
+     * The live fleet configures {@code exhaustedPattern: "The usage limit has been reached"} — with
+     * the leading {@code "The"}. Every other test here uses a pattern without it, which is the shape
+     * that made fleetd #348 need a looser rule than a start-of-line check. This pins the deployed
+     * shape as well, so a later tightening of {@link CompletionResolver} cannot silently stop
+     * recording the exhaustion this fleet actually reports.
+     *
+     * <p>What it does not prove: that this is the only pattern shape an operator will write.
+     */
+    @Test
+    void anExhaustionPatternCarryingItsLeadingWordsStillNotifiesTheSink() {
+        String block = "⏺ │ The usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("The usage limit has been reached");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
+                "the live pattern shape must still fail the send as exhausted");
+        assertEquals(1, notified.size(),
+                "the live pattern shape must still reach the sink");
+    }
+
     @Test
     void aLosingBackendExhaustedClassificationNeverNotifiesTheExhaustionSink() {
         // The waiter was already resolved (e.g. by the worker's own reply) before this scrape landed —
@@ -804,7 +872,28 @@ class CompletionResolverTest {
     // --- fleetd#201 Unit 1: target-keyed backend-error pattern + typed sink ----------------------
 
     @Test
-    void aConfiguredBackendErrorPatternClassifiesAMatchAsAFailureAndNotifiesTheSinkOnce() {
+    void aNormalMemberReportMentioningTheFallbackErrorPatternFailsButDoesNotNotifyTheSink() {
+        String block = "⏺ I checked the retry path. An API Error: makes it back off.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        BackendErrorSink sink = (target, matchedLine, reason) -> notified.add(target + ": " + matchedLine);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), BackendErrorPatternLookup.legacy(), sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "a scrape mentioning the pattern must still fail the send");
+        assertTrue(waiter.getNow(null).text().contains("I checked the retry path. An API Error: makes it back off."),
+                "the failure must keep the whole pane tail");
+        assertTrue(notified.isEmpty(),
+                "a normal report mentioning the fallback pattern must not record a credential failure");
+    }
+
+    @Test
+    void aConfiguredBackendErrorPatternAtTheStartOfALineClassifiesAMatchAndNotifiesTheSinkOnce() {
         String block = "⏺ 503 Service Unavailable: upstream credential rejected\n❯ ";
         FakeHerdr herdr = new FakeHerdr().readText(block);
         Rendezvous rendezvous = new Rendezvous();
@@ -924,7 +1013,7 @@ class CompletionResolverTest {
     }
 
     @Test
-    void aConfiguredPatternAlsoClassifiesTheRawScrapeFallbackAndNotifiesTheSink() {
+    void aConfiguredPatternAtTheStartOfALineAlsoClassifiesTheRawScrapeFallbackAndNotifiesTheSink() {
         // No ⏺ marker and leading TUI chrome ⇒ lastAssistantBlock() yields "", so classification must
         // fall back to the raw scrape (fleetd#211) — and it must use the configured pattern too.
         String block = """
@@ -949,6 +1038,40 @@ class CompletionResolverTest {
         assertTrue(notified.get(0).contains("503 Service Unavailable"), notified.get(0));
     }
 
+    /**
+     * fleetd #339 follow-up: a genuine backend error rendered behind terminal chrome must still
+     * record the credential outage. #339 added a start-of-line check to stop a member's own prose
+     * being counted as an outage, and a bare {@code lookingAt} also rejected this — the send failed
+     * but the sink never fired. #339's invariant 3 named that direction as the worse one: a real
+     * outage going unrecorded leaves the fleet spawning into a dead credential.
+     *
+     * <p>The raw-scrape path is where this matters, because its own comment says to expect leading
+     * TUI chrome there.
+     */
+    @Test
+    void aRealErrorBehindTerminalChromeStillNotifiesTheSink() {
+        String block = """
+                ╭──────────────────────────────────────╮
+                │ 503 Service Unavailable: upstream credential rejected
+                """;
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        BackendErrorPatternLookup patterns = target -> Pattern.compile("(?i)503 Service Unavailable");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        BackendErrorSink sink = (target, matchedLine, reason) -> notified.add(target + ": " + matchedLine);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "a real backend error must still fail the send");
+        assertEquals(1, notified.size(),
+                "a real error line behind box chrome is still a real outage — it must reach the sink, "
+                        + "or the fleet keeps spawning into a dead credential");
+    }
+
     // --- fleetd#201 Unit 1: classification inside the fleetd#164 MIN_TURN_NANOS floor -------------
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java

@@ -48,7 +48,7 @@ class InjectorTest {
 
     @Test
     void deliversWhenIdle() {
-        CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
+        CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T)).completion();
         assertFalse(f.isDone(), "not delivered until an injectable status arrives");
         injector.onStatus(T, AgentStatus.IDLE);
         assertTrue(f.isDone());
@@ -170,6 +170,32 @@ class InjectorTest {
         assertEquals(List.of("a", "b", "c"), sent());
     }
 
+    @Test
+    void cancellingTheMiddleDeliveryKeepsTheFollowingDeliveryReachable() {
+        Injector.Delivery first = injector.enqueue(T, "same text", TestTurnTokens.inert(T));
+        Injector.Delivery cancelled = injector.enqueue(T, "same text", TestTurnTokens.inert(T));
+        injector.enqueue(T, "after cancelled", TestTurnTokens.inert(T));
+
+        assertEquals(Injector.Cancellation.CANCELLED, injector.cancel(cancelled));
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+        injector.onStatus(T, AgentStatus.IDLE);
+
+        assertEquals(List.of("same text", "after cancelled"), sent(),
+                "cancellation must match the exact Delivery and preserve the remaining FIFO queue");
+        assertTrue(first.completion().isDone());
+    }
+
+    @Test
+    void cancellationReportsDeliveredWhenPickupWonTheRace() {
+        Injector.Delivery delivery = injector.enqueue(T, "already sent", TestTurnTokens.inert(T));
+        injector.onStatus(T, AgentStatus.IDLE);
+
+        assertEquals(Injector.Cancellation.DELIVERED, injector.cancel(delivery),
+                "a cancellation after pickup must not claim that the text stayed queued");
+        assertEquals(List.of("already sent"), sent());
+    }
+
     @Test
     void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
         assertTrue(injector.activeTargets().isEmpty());
@@ -378,7 +404,7 @@ class InjectorTest {
     void sendFailureDropsMessageAndFailsItsFuture() {
         FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
         Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T)).completion();
 
         inj.onStatus(T, AgentStatus.IDLE);
         assertTrue(f.isCompletedExceptionally());
@@ -387,7 +413,7 @@ class InjectorTest {
 
     @Test
     void dropFailsPendingWaiters() {
-        CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
+        CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T)).completion();
         injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
         assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
     }
@@ -396,8 +422,8 @@ class InjectorTest {
     void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
         Captor cap = new Captor();
         Injector inj = new Injector(new AgentControl(herdr), cap);
-        CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
-        CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));
+        CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T)).completion();
+        CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T)).completion();
 
         inj.onStatus(T, AgentStatus.IDLE);    // deliver the first message
         inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
@@ -449,7 +475,7 @@ class InjectorTest {
         Captor cap = new Captor();
         List<String> forgotten = new ArrayList<>();
         Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
-        CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));
+        CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T)).completion();
 
         for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
 
@@ -563,7 +589,7 @@ class InjectorTest {
         StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
         poller.start();
         try {
-            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
+            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T)).completion();
             delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
         } finally {
             poller.stop();
@@ -582,7 +608,7 @@ class InjectorTest {
     void deliveredFutureCarriesSendFailure() {
         FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
         Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T)).completion();
         inj.onStatus(T, AgentStatus.IDLE);
         ExecutionException ex = assertThrows(ExecutionException.class, f::get);
         assertInstanceOf(HerdrException.class, ex.getCause());

fleetd/src/test/java/dev/ltms/fleet/inject/StatusPollerRoutingTest.java

@@ -41,7 +41,7 @@ class StatusPollerRoutingTest {
         poller.start();
         try {
             CompletableFuture<Void> delivered =
-                    injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET));
+                    injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET)).completion();
             // Must resolve quickly: refining against the WRONG daemon (member) never classifies
             // out of UNKNOWN, so this would time out under the bug.
             delivered.get(2, TimeUnit.SECONDS);
@@ -65,7 +65,7 @@ class StatusPollerRoutingTest {
         poller.start();
         try {
             CompletableFuture<Void> delivered =
-                    injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET));
+                    injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET)).completion();
             assertThrows(TimeoutException.class, () -> delivered.get(500, TimeUnit.MILLISECONDS),
                     "a lead target must never be refined from the member daemon's pane content");
         } finally {

fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java

@@ -299,6 +299,41 @@ class CompositePeerLauncherTest {
                 "two adapter kinds sharing one daemon keep the fallback route");
     }
 
+    @Test
+    void stopThroughTheSingleDaemonShortcutStillClosesTheTabWhenTheFallbackDelegateUsesPanePlacement() {
+        // fleetd #342: the single-daemon shortcut (spawnedBy empty, herdrDaemonCount()==1) always
+        // routes stop() through delegates.getFirst() — here the claude adapter, configured for
+        // PANE placement (its own profiles never create a dedicated tab). The pane being torn
+        // down here actually belongs to the opencode adapter's TAB placement, sharing the same
+        // herdr daemon — mixing placements is the point: every existing stop-fallback test in this
+        // class configures BOTH adapters as "tab", so usesTabPlacement() was true either way and
+        // the mis-routing never showed.
+        //
+        // Before the fix, HerdrPeerLauncher#stop gated tab resolution on usesTabPlacement() of the
+        // delegate it happened to be called through, so the wrongly-routed (pane-placement) claude
+        // adapter never even looked for a tab to close, and the now-empty tab leaked with nothing
+        // to reap it. The fix (fleetd #342) resolves the pane's real tab unconditionally, so the
+        // decision follows the pane's actual placement rather than the fallback delegate's static
+        // config.
+        FakeHerdr herdr = new FakeHerdr();
+        FleetConfig.Profile claudePane = new FleetConfig.Profile("claude", "http://gx00.gw:8000",
+                "coder", null, "FLEETD_WORKER_TOKEN", List.of("claude"), "pane", "fleetd-workers",
+                "w #{n}", null, null, null);
+        ClaudeCodeLauncher claude = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of("claude", claudePane), "claude", _ -> null);
+        PeerLauncher composite = new CompositePeerLauncher(List.of(claude, opencodeAdapter(herdr)), "claude");
+
+        // "w9:pW" was never spawned through this composite instance, so spawnedBy has no entry for
+        // it (the same in-memory-cache-miss shape a daemon restart leaves behind) — stop() falls
+        // through to the single-daemon shortcut and hands it to delegates.getFirst() (claude).
+        composite.stop("w9:pW");
+
+        assertTrue(herdr.called("pane.close"), "the pane itself is still closed on every routed path");
+        assertTrue(herdr.called("tab.close"),
+                "the pane's real (sole-occupant) tab must be closed even though the fallback routed "
+                        + "through a delegate configured for pane placement");
+    }
+
     @Test
     void listKeepsBothPanesWhenTwoDaemonsShareAPaneId() {
         // herdr pane ids are per-daemon counters, so two daemons really can both hold w1:p1 on

fleetd/src/test/java/dev/ltms/fleet/member/HerdrPeerLauncherAllowListWiringTest.java

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -370,6 +371,161 @@ class HerdrPeerLauncherAllowListWiringTest {
                 "expected the pre-existing 'scrub blanks them' INFO unchanged, got: " + messages);
     }
 
+    /**
+     * fleetd #341: {@code unprotectedGapLogged} guarded TWO WARN branches that name DIFFERENT env
+     * var names — the allow-list branch ({@code keptByDerivedList}, below) and the deny-by-default
+     * / non-zsh-fallback branch ({@link HerdrPeerLauncher#warnGapUnprotected}). {@code
+     * memberCredentials} is a live, re-read-per-spawn supplier, so the policy can change between
+     * two spawns on the same launcher instance — a config reload needs no restart. Spawn 1 runs
+     * under {@code deny-by-default} with a gap of {@code SPAWN_ONE_UNCOVERED_TOKEN}, which trips
+     * the (before this fix) SHARED one-shot flag. The policy is then reloaded to {@code
+     * allow-list}; spawn 2's gap is {@code FLEETD_WORKER_TOKEN} instead — the test profile's own
+     * {@code tokenEnv}, which the derived allow-list keeps even though it is on neither {@code
+     * known:} nor {@code allow:}, so it is genuinely unprotected and deserves its own WARN. Before
+     * this fix that WARN never fires, because the shared flag was already {@code true} — the
+     * operator is never told {@code FLEETD_WORKER_TOKEN} reaches every member pane unblocked. Real
+     * path: two real {@link HerdrPeerLauncher#spawn} calls on ONE launcher instance, with mutable
+     * {@code memberCredentials}/host-env suppliers standing in for a live config reload between
+     * spawns.
+     */
+    @Test
+    void aDifferentUnprotectedGapOnALaterSpawnIsNotSuppressedByAnEarlierSpawnsWarn() {
+        FakeHerdr herdr = new FakeHerdr();
+        AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
+                new FleetConfig.MemberCredentials(null, List.of(), List.of(), null)); // deny-by-default
+        AtomicReference<Set<String>> hostEnvState =
+                new AtomicReference<>(Set.of("SPAWN_ONE_UNCOVERED_TOKEN"));
+        WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
+
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        Level original = logger.getLevel();
+        logger.setLevel(Level.WARN);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            // Spawn 1: deny-by-default, gap = {SPAWN_ONE_UNCOVERED_TOKEN} — the effectiveAllowed ==
+            // null branch, via warnGapUnprotected.
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+
+            // Live policy reload to allow-list, with a DIFFERENT gap name.
+            credsState.set(new FleetConfig.MemberCredentials(
+                    FleetConfig.MemberCredentials.POLICY_ALLOW_LIST, List.of(), List.of(), null));
+            hostEnvState.set(Set.of("FLEETD_WORKER_TOKEN"));
+            // Spawn 2: allow-list, gap = {FLEETD_WORKER_TOKEN} — kept by the derived allow-list
+            // (the profile's own tokenEnv), so it is the effectiveAllowed != null / keptByDerivedList
+            // branch, at the SAME log line HerdrPeerLauncher:1819 guards with the shared flag.
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(original);
+        }
+
+        List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
+        assertTrue(messages.stream().anyMatch(
+                        m -> m.contains("UNBLOCKED") && m.contains("SPAWN_ONE_UNCOVERED_TOKEN")),
+                "spawn 1's deny-by-default gap must still warn — got: " + messages);
+        assertTrue(messages.stream().anyMatch(
+                        m -> m.contains("UNBLOCKED") && m.contains("FLEETD_WORKER_TOKEN")),
+                "spawn 2's gap names a DIFFERENT env var than spawn 1 (FLEETD_WORKER_TOKEN, not "
+                        + "SPAWN_ONE_UNCOVERED_TOKEN) — it must still be warned about even though a "
+                        + "flag already fired once for spawn 1's unrelated name. Before fleetd #341's "
+                        + "fix this WARN never fires because unprotectedGapLogged was already true. "
+                        + "Got: " + messages);
+    }
+
+    /**
+     * fleetd #341 follow-up: the OTHER half of the guard's contract. The set exists to report every
+     * distinct name, but it must still report each one only ONCE — the noise control is the reason
+     * a guard is here at all, and the ticket named it as invariant 1. Two spawns, same policy, same
+     * gap name: exactly one WARN mentioning it.
+     *
+     * <p>Measured before this test existed: replacing {@code .filter(unprotectedGapNamesWarned::add)}
+     * with a filter that adds and always returns {@code true} — so every name is logged on every
+     * spawn — left all 1358 tests green. The fix was correct and nothing held it there. That is the
+     * "a test on the seam does not prove the caller" shape: the {@code Set} behaves, and nothing
+     * proved this class used it as a guard rather than as a record.
+     */
+    @Test
+    void theSameUnprotectedNameIsWarnedAboutOnlyOnceAcrossSpawns() {
+        FakeHerdr herdr = new FakeHerdr();
+        AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
+                new FleetConfig.MemberCredentials(null, List.of(), List.of(), null)); // deny-by-default
+        AtomicReference<Set<String>> hostEnvState =
+                new AtomicReference<>(Set.of("REPEATED_UNCOVERED_TOKEN"));
+        WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
+
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        Level original = logger.getLevel();
+        logger.setLevel(Level.WARN);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+            // Same policy, same gap, second spawn. Nothing new to tell the operator.
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(original);
+        }
+
+        List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
+        long mentioning = messages.stream()
+                .filter(m -> m.contains("REPEATED_UNCOVERED_TOKEN"))
+                .count();
+        assertEquals(1, mentioning,
+                "one unchanged unprotected name across two spawns must produce exactly one WARN — "
+                        + "the set is a guard, not just a record. Got: " + messages);
+    }
+
+    /**
+     * fleetd #341 follow-up: the reverse policy order. The original defect was found going
+     * deny-by-default then allow-list, and a guard that is fixed in one direction is not
+     * necessarily fixed in the other — "ask which states still OPEN the gate". Here spawn 1 runs
+     * under {@code allow-list} (the {@code keptByDerivedList} branch) and spawn 2 under
+     * {@code deny-by-default} ({@link HerdrPeerLauncher#warnGapUnprotected}), with a different name
+     * each time. Both must be reported.
+     */
+    @Test
+    void anAllowListWarnDoesNotSuppressALaterDenyByDefaultWarnForADifferentName() {
+        FakeHerdr herdr = new FakeHerdr();
+        AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
+                new FleetConfig.MemberCredentials(
+                        FleetConfig.MemberCredentials.POLICY_ALLOW_LIST, List.of(), List.of(), null));
+        AtomicReference<Set<String>> hostEnvState =
+                new AtomicReference<>(Set.of("FLEETD_WORKER_TOKEN"));
+        WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
+
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        Level original = logger.getLevel();
+        logger.setLevel(Level.WARN);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            // Spawn 1: allow-list, gap kept by the derived list — the keptByDerivedList WARN.
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+
+            // Live reload the OTHER way: back to deny-by-default, with a different name.
+            credsState.set(new FleetConfig.MemberCredentials(null, List.of(), List.of(), null));
+            hostEnvState.set(Set.of("LATER_UNCOVERED_TOKEN"));
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(original);
+        }
+
+        List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
+        assertTrue(messages.stream().anyMatch(
+                        m -> m.contains("UNBLOCKED") && m.contains("FLEETD_WORKER_TOKEN")),
+                "spawn 1's allow-list gap must warn — got: " + messages);
+        assertTrue(messages.stream().anyMatch(
+                        m -> m.contains("UNBLOCKED") && m.contains("LATER_UNCOVERED_TOKEN")),
+                "spawn 2's deny-by-default gap names a different variable and must still be warned "
+                        + "about, even though an allow-list WARN already fired. Got: " + messages);
+    }
+
     /**
      * fleetd #185 stage 2: with {@code memberHerdrSocket:} configured, member panes run under a
      * different OS user — {@link HerdrPeerLauncher#hostEnvNames} describes fleetd's own process, not

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

@@ -0,0 +1,134 @@
+package dev.ltms.fleet.msg;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AmqpConnectionFailureLoggerTest {
+
+    @Test
+    void installedHandlersLogTheirOwnConnectionNamesAtErrorWithTheCause() throws Exception {
+        ConnectionFactory inboxFactory = AmqpReplyInbox.connectionFactory("amqp://127.0.0.1");
+        ConnectionFactory mailboxFactory = LeadMailbox.connectionFactory("amqp://127.0.0.1");
+
+        AmqpConnectionFailureLogger inboxHandler = installedStrictHandler(inboxFactory, "reply inbox");
+        AmqpConnectionFailureLogger mailboxHandler = installedStrictHandler(mailboxFactory, "lead mailbox");
+        assertEquals(AmqpConnectionFailureLogger.REPLY_INBOX, inboxHandler.connectionName());
+        assertEquals(AmqpConnectionFailureLogger.LEAD_MAILBOX, mailboxHandler.connectionName());
+
+        ListAppender<ILoggingEvent> inboxEvents = attach(AmqpReplyInbox.class);
+        ListAppender<ILoggingEvent> mailboxEvents = attach(LeadMailbox.class);
+        IllegalStateException inboxFailure = new IllegalStateException("inbox failure");
+        IllegalStateException mailboxFailure = new IllegalStateException("mailbox failure");
+        try {
+            inboxHandler.handleUnexpectedConnectionDriverException(null, inboxFailure);
+            mailboxHandler.handleConnectionRecoveryException(null, mailboxFailure);
+
+            assertError(inboxEvents, "AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred",
+                    inboxFailure, "inbox failure line");
+            assertError(mailboxEvents, "AMQP connection fleetd-lead-mailbox: Caught an exception during connection recovery!",
+                    mailboxFailure, "mailbox recovery line");
+        } finally {
+            detach(AmqpReplyInbox.class, inboxEvents);
+            detach(LeadMailbox.class, mailboxEvents);
+        }
+    }
+
+    @Test
+    void connectionResetKeepsForgivingHandlerWarningSemantics() {
+        AmqpConnectionFailureLogger handler = new AmqpConnectionFailureLogger(
+                AmqpConnectionFailureLogger.REPLY_INBOX, LoggerFactory.getLogger(AmqpReplyInbox.class));
+        ListAppender<ILoggingEvent> events = attach(AmqpReplyInbox.class);
+        try {
+            handler.handleUnexpectedConnectionDriverException(null, new IOException("Connection reset"));
+            assertEquals(1, events.list.size(), "the handler must still log a reset");
+            ILoggingEvent event = events.list.getFirst();
+            assertEquals(Level.WARN, event.getLevel(), "ForgivingExceptionHandler logs connection resets at WARN");
+            assertEquals("AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred "
+                            + "(Exception message: Connection reset)", event.getFormattedMessage());
+            assertTrue(event.getThrowableProxy() == null, "ForgivingExceptionHandler does not attach a reset stack trace");
+        } finally {
+            detach(AmqpReplyInbox.class, events);
+        }
+    }
+
+    @Test
+    void connectionNamesStayDistinct() {
+        assertNotEquals(AmqpConnectionFailureLogger.REPLY_INBOX, AmqpConnectionFailureLogger.LEAD_MAILBOX,
+                "reply-inbox and lead-mailbox failures must be distinguishable");
+    }
+
+    @Test
+    void strictConsumerExceptionStillClosesItsChannel() {
+        AtomicInteger closes = new AtomicInteger();
+        Channel channel = (Channel) Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] {Channel.class},
+                (_, method, _) -> switch (method.getName()) {
+                    case "close" -> {
+                        closes.incrementAndGet();
+                        yield null;
+                    }
+                    case "toString" -> "test-channel";
+                    default -> throw new UnsupportedOperationException(method.getName());
+                });
+        AmqpConnectionFailureLogger handler = new AmqpConnectionFailureLogger(
+                AmqpConnectionFailureLogger.REPLY_INBOX, LoggerFactory.getLogger(AmqpReplyInbox.class));
+
+        handler.handleConsumerException(channel, new IllegalStateException("consumer failed"), null, "tag", "handleDelivery");
+
+        assertEquals(1, closes.get(), "DefaultExceptionHandler must close a channel after a consumer exception");
+    }
+
+    @Test
+    void handlerOnlyChangesDefaultHandlerLogging() {
+        assertEquals(DefaultExceptionHandler.class,
+                AmqpConnectionFailureLogger.class.getSuperclass());
+        assertFalse(java.util.Arrays.stream(AmqpConnectionFailureLogger.class.getDeclaredMethods())
+                        .anyMatch(method -> method.getName().startsWith("handle")),
+                "all exception-handling methods must remain inherited from DefaultExceptionHandler");
+    }
+
+    private static ListAppender<ILoggingEvent> attach(Class<?> owner) {
+        Logger logger = (Logger) LoggerFactory.getLogger(owner);
+        logger.setLevel(Level.DEBUG);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        return appender;
+    }
+
+    private static void detach(Class<?> owner, ListAppender<ILoggingEvent> appender) {
+        ((Logger) LoggerFactory.getLogger(owner)).detachAppender(appender);
+    }
+
+    private static AmqpConnectionFailureLogger installedStrictHandler(ConnectionFactory factory, String connection) {
+        assertInstanceOf(DefaultExceptionHandler.class, factory.getExceptionHandler(),
+                connection + " must keep DefaultExceptionHandler: replacing the strict handler with a forgiving one "
+                        + "changes when a channel is closed");
+        return assertInstanceOf(AmqpConnectionFailureLogger.class, factory.getExceptionHandler());
+    }
+
+    private static void assertError(ListAppender<ILoggingEvent> events, String message, Throwable cause, String name) {
+        assertEquals(1, events.list.size(), name);
+        ILoggingEvent event = events.list.getFirst();
+        assertEquals(Level.ERROR, event.getLevel(), name);
+        assertEquals(message, event.getFormattedMessage(), name);
+        assertEquals(cause.toString(), event.getThrowableProxy().getClassName() + ": "
+                + event.getThrowableProxy().getMessage(), name);
+    }
+}

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

@@ -258,7 +258,7 @@ class MessageServiceTest {
         injector.onStatus(T, AgentStatus.IDLE);    // first delivery
         injector.onStatus(T, AgentStatus.WORKING); // first turn in flight
 
-        CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
+        CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T)).completion();
         CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
         injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
 
@@ -377,6 +377,126 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.QUESTION, q.outcome());
     }
 
+    /**
+     * fleetd #334. {@code ask()}'s {@code TimeoutException} catch used to forget this task's
+     * {@code turnId} mapping ({@code clearAsyncQuestion(turnId, true)}) BEFORE closing the ask
+     * ({@code rendezvous.closeAsk}, in the shared {@code finally}). A primary's {@code answer()}
+     * call racing that exact window found the ask still "answerable" ({@code
+     * rendezvous.askSession(turnId)} still non-null) while the {@code Task} was already forgotten,
+     * so its {@code task != null} guard skipped the completion and the async ticket sat at
+     * {@code PENDING} forever even though {@code answer()} itself reported a result. The fix
+     * (closing the ask first) makes this window impossible: a racing {@code answer()} call either
+     * still finds the ask open (and the {@code Task} mapping guaranteed intact) or finds it already
+     * closed (and bails {@code STALE_TURN} before ever touching the {@code Task}). This test pins
+     * the exact window with {@code askTimeoutRaceHookForTest} and proves both invariants the ticket
+     * named: (1) a late/racing {@code answer()} sees the ask as already lapsed ({@code STALE_TURN}),
+     * never made answerable again, and (2) the async ticket still resolves {@code DONE} once the
+     * worker's real {@code fleet_reply} lands — it is never stranded {@code PENDING}.
+     */
+    /**
+     * fleetd #334 gated the ask-timeout teardown on {@code ticket.fresh()}, matching the {@code
+     * finally} block that already did. This pins that gate. A coalesced duplicate passes its own
+     * {@code timeoutMillis}, which says nothing about whether the shared ask is done — so a
+     * duplicate timing out first must leave the fresh owner's still-open ask answerable.
+     *
+     * <p>Measured on merge: without this test, removing the {@code ticket.fresh()} gate left all
+     * 1371 tests green. The gate shipped with the reorder and nothing held it there.
+     *
+     * <p>What this does not prove: anything about the ordering inside the gate — that is
+     * {@code aLateAnswerDuringAskTimeoutTeardownStillCompletesTheAsyncTicket}'s job.
+     */
+    @Test
+    void aCoalescedDuplicateAskTimingOutLeavesTheFreshOwnersAskOpen() throws Exception {
+        String ticket = messages.sendAsync(T, "long task");
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker picks it up, then pauses to ask
+
+        CompletableFuture<MessageService.AskResult> fresh =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+
+        MessageService.TaskView asking = null;
+        long deadline = System.currentTimeMillis() + 2000;
+        while ((asking == null || asking.phase() != MessageService.Phase.ASKING)
+                && System.currentTimeMillis() < deadline) {
+            asking = messages.poll(ticket);
+            //noinspection BusyWait
+            Thread.sleep(5);
+        }
+        assertNotNull(asking, "the fresh owner's question must surface before the duplicate asks");
+        String turnId = asking.turnId();
+        assertNotNull(turnId, "an ASKING view carries the turnId to answer on");
+
+        // A coalesced duplicate on the same session, with its own much shorter timeout.
+        MessageService.AskResult duplicate = messages.ask(T, "which config file?", 100);
+        assertEquals(MessageService.AskOutcome.TIMED_OUT, duplicate.outcome(),
+                "the duplicate's own timeout elapses first");
+
+        CompletableFuture<MessageService.Reply> answered =
+                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "fleetd.yaml", 500));
+
+        MessageService.AskResult a = fresh.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.AskOutcome.ANSWERED, a.outcome(),
+                "a duplicate's timeout must not lapse the ask the fresh owner still holds");
+        assertEquals("fleetd.yaml", a.answer());
+        assertFalse(answered.get(5, TimeUnit.SECONDS).outcome() == MessageService.Outcome.STALE_TURN,
+                "the answer must not be rejected as stale");
+    }
+
+    @Test
+    void aLateAnswerDuringAskTimeoutTeardownStillCompletesTheAsyncTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "long task");
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker picks it up, then pauses to ask
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 200));
+
+        // Wait for the question to actually surface (poll sees ASKING) before racing the timeout.
+        MessageService.TaskView asking = null;
+        long deadline = System.currentTimeMillis() + 2000;
+        while ((asking == null || asking.phase() != MessageService.Phase.ASKING)
+                && System.currentTimeMillis() < deadline) {
+            asking = messages.poll(ticket);
+            //noinspection BusyWait
+            Thread.sleep(5);
+        }
+        assertNotNull(asking, "the question must surface before the ask times out");
+        String turnId = asking.turnId();
+        assertNotNull(turnId, "an ASKING view carries the turnId to answer on");
+
+        CompletableFuture<MessageService.Reply> lateAnswer = new CompletableFuture<>();
+        messages.setAskTimeoutRaceHookForTest(() ->
+                lateAnswer.complete(messages.answer(turnId, "too late", 500)));
+        try {
+            MessageService.AskResult a = ask.get(5, TimeUnit.SECONDS);
+            assertEquals(MessageService.AskOutcome.TIMED_OUT, a.outcome());
+
+            MessageService.Reply late = lateAnswer.get(5, TimeUnit.SECONDS);
+            assertEquals(MessageService.Outcome.STALE_TURN, late.outcome(),
+                    "a late answer racing the timeout teardown must see the ask as already lapsed");
+
+            // The worker resumes on its own (per the ask() contract) and eventually sends its real
+            // fleet_reply; the async ticket must still resolve with it, not strand at PENDING.
+            assertTrue(messages.reply(T, "real result"), "the worker's real reply must still be accepted");
+        } finally {
+            messages.setAskTimeoutRaceHookForTest(null);
+        }
+
+        MessageService.TaskView done = null;
+        deadline = System.currentTimeMillis() + 2000;
+        while ((done == null || done.phase() == MessageService.Phase.PENDING)
+                && System.currentTimeMillis() < deadline) {
+            done = messages.poll(ticket);
+            //noinspection BusyWait
+            Thread.sleep(5);
+        }
+        assertNotNull(done);
+        assertEquals(MessageService.Phase.DONE, done.phase(), "the async ticket must not be stranded PENDING");
+        assertEquals("real result", done.reply());
+    }
+
     @Test
     void answeringAnUnknownTurnIsStale() {
         MessageService.Reply r = messages.answer(T + "#999", "too late", 500);
@@ -410,6 +530,29 @@ class MessageServiceTest {
                 "a delivered send whose worker never replies times out as still working");
     }
 
+    /**
+     * fleetd #345. This forces the injector to pick up the exact pending delivery after {@code send}
+     * first observes its completion as incomplete, but before {@code cancel} takes the target monitor.
+     * The timeout must use {@link Injector.Cancellation#DELIVERED} from {@code cancel} and report
+     * {@link MessageService.Outcome#TIMED_OUT_WORKING}, because the text landed.
+     *
+     * <p>What this does not prove: that this precise interleaving happens by itself under production
+     * timing. The test forces it through a test-only hook; it proves the timeout caller handles the
+     * injector result when the interleaving occurs.
+     */
+    @Test
+    void sendTimeoutUsesCancellationDeliveredWhenPickupWinsTheRace() {
+        messages.setTimeoutCancellationRaceHookForTest(() -> injector.onStatus(T, AgentStatus.IDLE));
+        try {
+            MessageService.Reply reply = messages.send(T, "race delivery", 50);
+
+            assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, reply.outcome(),
+                    "cancel reporting DELIVERED means the worker received the timed-out message");
+        } finally {
+            messages.setTimeoutCancellationRaceHookForTest(null);
+        }
+    }
+
     @Test
     void answerTimesOutWhenTheResumedWorkerNeverReplies() throws Exception {
         CompletableFuture<MessageService.Reply> send = sendAsync();
@@ -785,6 +928,49 @@ class MessageServiceTest {
         assertFailedTicket(third, "agent target term_a not found");
     }
 
+    // --- fleetd #335 (site 1): a per-task cleanup failure inside the abandon() loop must not -----
+    // strand the tasks that come after it. abandon()'s own comment on the loop documents the one
+    // real production call that can throw there (inbox.publish, in the stranded-reply put-back
+    // branch, reached when a concurrent reply() or a second abandon() races this one) — but
+    // reaching that branch requires the exact combination the #137 follow-up above already found
+    // unreachable through the public API. abandonCleanupHookForTest reproduces the resulting SHAPE
+    // (one task's cleanup throws) directly instead, the same technique this file already uses for
+    // fleetd #324/#329's own hard-to-time races.
+    @Test
+    void aPerTaskCleanupFailureDoesNotStrandTheRemainingMatchingTasks() throws Exception {
+        ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
+        try {
+            String first = messages.sendAsync(T, "first task");
+            awaitWaiting(); // first task owns the target lock and rendezvous waiter
+            String second = messages.sendAsync(T, "second task"); // parked on the same lock
+            String third = messages.sendAsync(T, "third task"); // parked too — the whole sweep must survive
+
+            java.util.concurrent.atomic.AtomicInteger calls = new java.util.concurrent.atomic.AtomicInteger();
+            messages.setAbandonCleanupHookForTest(() -> {
+                if (calls.getAndIncrement() == 0) {
+                    throw new RuntimeException("PROBE-335-SITE1");
+                }
+            });
+
+            assertTrue(messages.abandon(T, "agent target term_a not found"));
+
+            // Every task in the loop still gets its own outcome — the one whose cleanup threw
+            // included — even though the loop had no way to know in advance which one that would be.
+            assertFailedTicket(first, "agent target term_a not found");
+            assertFailedTicket(second, "agent target term_a not found");
+            assertFailedTicket(third, "agent target term_a not found");
+
+            assertTrue(appender.list.stream().anyMatch(e ->
+                            e.getLevel() == Level.ERROR
+                                    && e.getThrowableProxy() != null
+                                    && "PROBE-335-SITE1".equals(e.getThrowableProxy().getMessage())),
+                    "a per-task cleanup failure must still reach the log, not vanish silently");
+        } finally {
+            messages.setAbandonCleanupHookForTest(null);
+            detachMessageServiceLog(appender);
+        }
+    }
+
     // --- #137 follow-up: abandon() must not guess when more than one task is open ---------------
     //
     // A test combining a genuine stranded reply (hasStrandedReply(T)==true) with two simultaneously
@@ -1472,6 +1658,42 @@ class MessageServiceTest {
         }
     }
 
+    // --- fleetd #335 (site 2): task.future.whenComplete's own returned stage is discarded, so an --
+    // uncaught throw from ReplyPushLoop.onTicketTerminal used to vanish with no log line and no
+    // metric. The real production trigger is the daemon's own shutdown sequence (Fleetd's shutdown
+    // hook): messages.close() only stops the async executor from taking NEW work — it does not
+    // cancel a send already in flight — while pushLoop.close() shuts its scheduler down immediately
+    // right after, so a ticket that completes in that narrow window has onTicketTerminal's own
+    // scheduler.schedule(...) throw a real RejectedExecutionException. Reproduced here by shutting
+    // the very same scheduler down before the ticket resolves — no test-only hook needed, this
+    // reachable path throws for real.
+    @Test
+    void aTicketTerminalPushFailureDoesNotVanishSilently() throws Exception {
+        ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
+        try (var wiring = wireWithPushLoop(1, 50)) {
+            String ticket = wiring.service().sendAsync(T, "long task");
+            awaitWaiting();
+            wiring.scheduler().shutdownNow(); // simulate pushLoop.close() racing an in-flight send
+            injectDelivery();
+            assertTrue(rendezvous.resolve(T, "async result"));
+
+            // The ticket's own outcome must be unaffected by the swallowed exception — finishAsyncTask
+            // completes task.future before whenComplete's action (and thus onTicketTerminal) ever runs.
+            MessageService.TaskView done = awaitTicketPhaseOn(wiring.service(), ticket, MessageService.Phase.DONE);
+            assertEquals("async result", done.reply());
+
+            assertTrue(appender.list.stream().anyMatch(e ->
+                            e.getLevel() == Level.ERROR
+                                    && e.getFormattedMessage().contains(ticket)
+                                    && e.getThrowableProxy() != null
+                                    && "java.util.concurrent.RejectedExecutionException"
+                                            .equals(e.getThrowableProxy().getClassName())),
+                    "onTicketTerminal throwing must still reach the log, not vanish silently");
+        } finally {
+            detachMessageServiceLog(appender);
+        }
+    }
+
     // --- CB-582: fleet_ask question-open nudges --------------------------------------------------
 
     @Test
@@ -1768,7 +1990,10 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, r.outcome());
 
         assertTrue(messages.hasQueuedDelivery(T),
-                "a TIMED_OUT_QUEUED send leaves the message still queued in the injector");
+                "a TIMED_OUT_QUEUED send still records the undelivered delivery for fleet health");
+        injector.onStatus(T, AgentStatus.IDLE);
+        assertTrue(herdr.calls.stream().noneMatch(c -> c.method().equals("agent.prompt")),
+                "a TIMED_OUT_QUEUED send must be cancelled, not delivered when the worker later goes idle");
     }
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/rest/FleetAppTest.java

@@ -711,13 +711,21 @@ class FleetAppTest {
 
     @Test
     void stopWorkerInPanePlacementClosesOnlyThePane() throws Exception {
-        FakeHerdr herdr = new FakeHerdr();
+        // fleetd #342: tab-cleanup resolution is no longer skipped based on a profile's declared
+        // placement — a stop() routed through the wrong delegate (e.g. CompositePeerLauncher's
+        // single-daemon shortcut after a daemon restart) could carry a placement config that says
+        // nothing true about how THIS pane was actually placed. So the pane's real tab is now
+        // always resolved, and the single-occupant check below is what protects a pane-placement
+        // peer's shared tab, exactly as it always protected a tab-placement one. Model that
+        // realistically: the peer's pane was split into an existing tab that already held another
+        // occupant, so the tab must never be closed.
+        FakeHerdr herdr = new FakeHerdr().withWorkerTabPaneCount(2);
         int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"), "pane");
 
         assertEquals(204, req(port, "DELETE", "/members/w9:pW").statusCode());
         assertTrue(herdr.called("pane.close"));
-        assertFalse(herdr.called("tab.close"), "pane placement owns no tab to close");
-        assertFalse(herdr.called("pane.get"), "no tab resolution in pane placement");
+        assertTrue(herdr.called("pane.get"), "tab resolution now always runs, regardless of placement");
+        assertFalse(herdr.called("tab.close"), "pane placement's shared tab must never be closed");
     }
 
     @Test

scripts/redeploy-fleetd.sh

@@ -116,6 +116,50 @@ check_log_path_matches_plist() {
   ok "log path check: script and plist agree ($resolved_out)"
 }
 
+# Classify ERROR lines in one fresh log region. AMQP failure messages now include the connection
+# name, so a recovery can clear only errors for its own connection. A candidate with neither name
+# remains unexplained: it must never be quieted by a recovery on the other connection.
+classify_amqp_connection_errors() {
+  local log_file="$1" line pending_inbox=0 pending_lead_mailbox=0
+  REDEPLOY_ERROR_COUNT=0
+  REDEPLOY_RECOVERED_AMQP_ERRORS=0
+  REDEPLOY_UNEXPLAINED_ERRORS=0
+
+  while IFS= read -r line || [ -n "$line" ]; do
+    case "$line" in
+      *' ERROR '*|*' SEVERE '*)
+        REDEPLOY_ERROR_COUNT=$((REDEPLOY_ERROR_COUNT + 1))
+        case "$line" in
+          *'AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred'*|*'AMQP connection fleetd-reply-inbox: Caught an exception during connection recovery!'*)
+            pending_inbox=$((pending_inbox + 1))
+            ;;
+          *'AMQP connection fleetd-lead-mailbox: An unexpected connection driver error occurred'*|*'AMQP connection fleetd-lead-mailbox: Caught an exception during connection recovery!'*)
+            pending_lead_mailbox=$((pending_lead_mailbox + 1))
+            ;;
+          *'AMQP connection'*'An unexpected connection driver error occurred'*|*'AMQP connection'*'Caught an exception during connection recovery!'*)
+            REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1))
+            ;;
+          *) REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1)) ;;
+        esac
+        ;;
+      *'AMQP connection recovered; cleared held replies for fresh redelivery'*)
+        if [ "$pending_inbox" -gt 0 ]; then
+          pending_inbox=$((pending_inbox - 1))
+          REDEPLOY_RECOVERED_AMQP_ERRORS=$((REDEPLOY_RECOVERED_AMQP_ERRORS + 1))
+        fi
+        ;;
+      *'AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery'*)
+        if [ "$pending_lead_mailbox" -gt 0 ]; then
+          pending_lead_mailbox=$((pending_lead_mailbox - 1))
+          REDEPLOY_RECOVERED_AMQP_ERRORS=$((REDEPLOY_RECOVERED_AMQP_ERRORS + 1))
+        fi
+        ;;
+    esac
+  done < "$log_file"
+
+  REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + pending_inbox + pending_lead_mailbox))
+}
+
 # CB-600: sourceable for testing. When this file is SOURCED (not executed) it stops here — nothing
 # below runs — so a test harness can `source` it to call check_log_path_matches_plist (or the
 # other pure helpers above) against a throwaway plist fixture without ever reaching the mutating
@@ -361,15 +405,21 @@ tail -n "+$((RESTART_MARK + 1))" "$OUT" 2>/dev/null \
   | grep -iE 'deferred|classification:|fleet health:|coverage' | tail -8 | sed 's/^/   /' \
   || echo "   (nothing reported)"
 
-# Errors since the restart, anchored to the marker so old noise cannot leak in.
-ERRS="$(tail -n "+$((RESTART_MARK + 1))" "$OUT" 2>/dev/null | grep -cE ' (ERROR|SEVERE) ' || true)"
+# Errors since the restart, anchored to the marker so old noise cannot leak in. Keep the fresh
+# region in a file because the classifier must preserve the order of errors and recoveries.
+FRESH_LOG="$(mktemp -t fleetd-fresh-log)"
+trap 'rm -f "$FRESH_LOG"' EXIT
+tail -n "+$((RESTART_MARK + 1))" "$OUT" > "$FRESH_LOG" 2>/dev/null || true
+classify_amqp_connection_errors "$FRESH_LOG"
 say "result"
 ok "pid $NEW_PID, jar $(jar_id)"
-if [ "${ERRS:-0}" -gt 0 ]; then
-  warn "$ERRS ERROR lines since restart:"
-  tail -n "+$((RESTART_MARK + 1))" "$OUT" | grep -E ' (ERROR|SEVERE) ' | tail -5 | sed 's/^/   /'
-else
+if [ "$REDEPLOY_ERROR_COUNT" -eq 0 ]; then
   ok "no ERROR lines since restart"
+elif [ "$REDEPLOY_UNEXPLAINED_ERRORS" -eq 0 ]; then
+  ok "$REDEPLOY_RECOVERED_AMQP_ERRORS AMQP connection reset ERROR lines recovered since restart"
+else
+  warn "$REDEPLOY_ERROR_COUNT ERROR lines since restart:"
+  grep -E ' (ERROR|SEVERE) ' "$FRESH_LOG" | tail -5 | sed 's/^/   /'
 fi
 echo
 echo "   Next: call fleet_whoami and confirm it still answers 'primary'. A lead whose tab label"

scripts/test-redeploy-fleetd.sh

@@ -0,0 +1,238 @@
+#!/usr/bin/env bash
+# Self-contained checks for the pure log classifier in redeploy-fleetd.sh.
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+TMP="$(mktemp -d "$ROOT/.redeploy-log-test.XXXXXX")"
+trap 'rm -rf "$TMP"' EXIT
+
+# Sourcing stops before redeploy-fleetd.sh can build, stop, or start the daemon.
+source "$ROOT/scripts/redeploy-fleetd.sh"
+
+fail() {
+  printf 'FAIL: %s\n' "$*" >&2
+  return 1
+}
+
+assert_equals() {
+  local expected="$1" actual="$2" description="$3"
+  [ "$expected" = "$actual" ] || fail "$description: expected $expected, got $actual"
+}
+
+classify_fixture() {
+  local name="$1"
+  classify_amqp_connection_errors "$TMP/$name"
+}
+
+test_no_errors() {
+  cat > "$TMP/no-errors.log" <<'LOG'
+2026-09-05 12:00:00 INFO fleetd listening
+LOG
+  classify_fixture no-errors.log
+  assert_equals 0 "$REDEPLOY_ERROR_COUNT" "no-errors total"
+  assert_equals 0 "$REDEPLOY_UNEXPLAINED_ERRORS" "no-errors unexplained"
+}
+
+test_recovery_patterns_match_source() {
+  grep -F 'AMQP connection {}: {}' "$ROOT/fleetd/src/main/java/dev/ltms/fleet/msg/AmqpReplyInbox.java" > /dev/null \
+    || fail "AMQP failure pattern no longer matches source"
+  grep -F 'AMQP connection recovered; cleared held replies for fresh redelivery' \
+    "$ROOT/fleetd/src/main/java/dev/ltms/fleet/msg/AmqpReplyInbox.java" > /dev/null \
+    || fail "reply-inbox recovery pattern no longer matches source"
+  grep -F 'AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery' \
+    "$ROOT/fleetd/src/main/java/dev/ltms/fleet/msg/LeadMailbox.java" > /dev/null \
+    || fail "lead-mailbox recovery pattern no longer matches source"
+}
+
+test_attributed_recovered_connection_error() {
+  cat > "$TMP/attributed-recovered.log" <<'LOG'
+2026-09-05 12:00:00 ERROR [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+2026-09-05 12:00:01 INFO [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection recovered; cleared held replies for fresh redelivery
+LOG
+  classify_fixture attributed-recovered.log
+  assert_equals 1 "$REDEPLOY_ERROR_COUNT" "attributed-recovered total"
+  assert_equals 1 "$REDEPLOY_RECOVERED_AMQP_ERRORS" "attributed-recovered errors"
+  assert_equals 0 "$REDEPLOY_UNEXPLAINED_ERRORS" "attributed-recovered unexplained"
+}
+
+test_source_derived_error_shapes_recover_by_connection() {
+  # These ERROR shapes come from AmqpConnectionFailureLogger on main. They need a live-log check
+  # after redeploy because the new code has not yet written a production line.
+  cat > "$TMP/source-derived.log" <<'LOG'
+17:37:53.537 ERROR [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+17:37:54.537 ERROR [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection fleetd-reply-inbox: Caught an exception during connection recovery!
+17:37:55.537 ERROR [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+17:37:56.537 ERROR [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP connection fleetd-lead-mailbox: An unexpected connection driver error occurred
+17:37:57.537 ERROR [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP connection fleetd-lead-mailbox: Caught an exception during connection recovery!
+17:37:58.537 ERROR [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP connection fleetd-lead-mailbox: An unexpected connection driver error occurred
+17:38:00.000 INFO  [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection recovered; cleared held replies for fresh redelivery
+17:38:01.000 INFO  [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection recovered; cleared held replies for fresh redelivery
+17:38:02.000 INFO  [AMQP Connection broker:5672] d.l.fleet.msg.AmqpReplyInbox - AMQP connection recovered; cleared held replies for fresh redelivery
+17:38:03.000 INFO  [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+17:38:04.000 INFO  [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+17:38:05.000 INFO  [AMQP Connection broker:5672] d.ltms.fleet.msg.LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+LOG
+  classify_fixture source-derived.log
+  assert_equals 6 "$REDEPLOY_ERROR_COUNT" "source-derived total"
+  assert_equals 6 "$REDEPLOY_RECOVERED_AMQP_ERRORS" "source-derived recovered"
+  assert_equals 0 "$REDEPLOY_UNEXPLAINED_ERRORS" "source-derived unexplained"
+}
+
+test_cross_connection_unattributable_errors_stay_loud() {
+  # This candidate has neither stable connection name, so LeadMailbox recovery must not consume it.
+  cat > "$TMP/cross-unattributable.log" <<'LOG'
+2026-09-05 12:00:00 ERROR [AMQP Connection broker:5672] unknown - AMQP connection: An unexpected connection driver error occurred
+2026-09-05 12:00:01 ERROR [AMQP Connection broker:5672] unknown - AMQP connection: An unexpected connection driver error occurred
+2026-09-05 12:00:02 INFO  [AMQP Connection 10.10.20.13:5672] d.ltms.fleet.msg.LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+2026-09-05 12:00:03 INFO  [AMQP Connection 10.10.20.13:5672] d.ltms.fleet.msg.LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+LOG
+  classify_fixture cross-unattributable.log
+  assert_equals 2 "$REDEPLOY_ERROR_COUNT" "cross-unattributable total"
+  assert_equals 0 "$REDEPLOY_RECOVERED_AMQP_ERRORS" "cross-unattributable recovered"
+  assert_equals 2 "$REDEPLOY_UNEXPLAINED_ERRORS" "cross-unattributable unexplained"
+}
+
+test_attributed_cross_connection_errors_stay_loud() {
+  # LeadMailbox recovery cannot heal AmqpReplyInbox errors.
+  cat > "$TMP/cross-attributed.log" <<'LOG'
+2026-09-05 12:00:00 ERROR AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+2026-09-05 12:00:01 ERROR AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+2026-09-05 12:00:02 INFO LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+2026-09-05 12:00:03 INFO LeadMailbox - AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery
+LOG
+  classify_fixture cross-attributed.log
+  assert_equals 2 "$REDEPLOY_ERROR_COUNT" "cross-attributed total"
+  assert_equals 0 "$REDEPLOY_RECOVERED_AMQP_ERRORS" "cross-attributed recovered"
+  assert_equals 2 "$REDEPLOY_UNEXPLAINED_ERRORS" "cross-attributed unexplained"
+}
+
+test_attributed_unrecovered_connection_error() {
+  cat > "$TMP/unrecovered.log" <<'LOG'
+2026-09-05 12:00:00 ERROR AmqpReplyInbox - AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred
+LOG
+  classify_fixture unrecovered.log
+  assert_equals 1 "$REDEPLOY_ERROR_COUNT" "unrecovered total"
+  assert_equals 0 "$REDEPLOY_RECOVERED_AMQP_ERRORS" "unrecovered AMQP errors"
+  assert_equals 1 "$REDEPLOY_UNEXPLAINED_ERRORS" "unrecovered unexplained"
+}
+
+test_other_error_is_unexplained() {
+  cat > "$TMP/other-error.log" <<'LOG'
+2026-09-05 12:00:00 ERROR dev.ltms.fleet.Fleetd - startup failed
+2026-09-05 12:00:01 INFO dev.ltms.fleet.msg.AmqpReplyInbox - AMQP connection recovered; cleared held replies for fresh redelivery
+LOG
+  classify_fixture other-error.log
+  assert_equals 1 "$REDEPLOY_ERROR_COUNT" "other-error total"
+  assert_equals 1 "$REDEPLOY_UNEXPLAINED_ERRORS" "other-error unexplained"
+}
+
+test_recovery_requirement_mutation_is_caught() {
+  classify_amqp_connection_errors() {
+    local log_file="$1" line
+    REDEPLOY_ERROR_COUNT=0
+    REDEPLOY_RECOVERED_AMQP_ERRORS=0
+    REDEPLOY_UNEXPLAINED_ERRORS=0
+    while IFS= read -r line || [ -n "$line" ]; do
+      case "$line" in
+        *' ERROR '*|*' SEVERE '*)
+          REDEPLOY_ERROR_COUNT=$((REDEPLOY_ERROR_COUNT + 1))
+          case "$line" in
+            *'AMQP connection fleetd-reply-inbox: An unexpected connection driver error occurred'*)
+              REDEPLOY_RECOVERED_AMQP_ERRORS=$((REDEPLOY_RECOVERED_AMQP_ERRORS + 1))
+              ;;
+            *) REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1)) ;;
+          esac
+          ;;
+      esac
+    done < "$log_file"
+  }
+
+  if test_attributed_unrecovered_connection_error > "$TMP/mutation-output" 2>&1; then
+    fail "mutation accepted an unrecovered connection error"
+  fi
+  grep -F 'FAIL: unrecovered AMQP errors: expected 0, got 1' "$TMP/mutation-output" > /dev/null \
+    || fail "mutation failed without the expected assertion"
+  printf 'Recovery mutation: FAIL: unrecovered AMQP errors: expected 0, got 1\n'
+}
+
+test_shared_counter_mutation_is_caught() {
+  classify_amqp_connection_errors() {
+    local log_file="$1" line pending=0
+    REDEPLOY_ERROR_COUNT=0
+    REDEPLOY_RECOVERED_AMQP_ERRORS=0
+    REDEPLOY_UNEXPLAINED_ERRORS=0
+    while IFS= read -r line || [ -n "$line" ]; do
+      case "$line" in
+        *' ERROR '*|*' SEVERE '*)
+          REDEPLOY_ERROR_COUNT=$((REDEPLOY_ERROR_COUNT + 1))
+          case "$line" in
+            *'AMQP connection'*'An unexpected connection driver error occurred'*|*'AMQP connection'*'Caught an exception during connection recovery!'*)
+              case "$line" in
+                *'fleetd-reply-inbox'*|*'fleetd-lead-mailbox'*) pending=$((pending + 1)) ;;
+                *) REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1)) ;;
+              esac
+              ;;
+            *) REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1)) ;;
+          esac
+          ;;
+        *'AMQP connection recovered; cleared held replies for fresh redelivery'*|*'AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery'*)
+          if [ "$pending" -gt 0 ]; then
+            pending=$((pending - 1))
+            REDEPLOY_RECOVERED_AMQP_ERRORS=$((REDEPLOY_RECOVERED_AMQP_ERRORS + 1))
+          fi
+          ;;
+      esac
+    done < "$log_file"
+    REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + pending))
+  }
+
+  if test_attributed_cross_connection_errors_stay_loud > "$TMP/shared-mutation-output" 2>&1; then
+    fail "shared counter mutation accepted cross-connection recovery"
+  fi
+  grep -F 'FAIL: cross-attributed recovered: expected 0, got 2' "$TMP/shared-mutation-output" > /dev/null \
+    || fail "shared counter mutation failed without the expected assertion"
+  printf 'Shared-counter mutation: FAIL: cross-attributed recovered: expected 0, got 2\n'
+}
+
+test_unattributable_quiet_mutation_is_caught() {
+  classify_amqp_connection_errors() {
+    local log_file="$1" line
+    REDEPLOY_ERROR_COUNT=0
+    REDEPLOY_RECOVERED_AMQP_ERRORS=0
+    REDEPLOY_UNEXPLAINED_ERRORS=0
+    while IFS= read -r line || [ -n "$line" ]; do
+      case "$line" in
+        *' ERROR '*|*' SEVERE '*)
+          REDEPLOY_ERROR_COUNT=$((REDEPLOY_ERROR_COUNT + 1))
+          case "$line" in
+            *'AMQP connection'*'An unexpected connection driver error occurred'*|*'AMQP connection'*'Caught an exception during connection recovery!'*)
+              REDEPLOY_RECOVERED_AMQP_ERRORS=$((REDEPLOY_RECOVERED_AMQP_ERRORS + 1))
+              ;;
+            *) REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + 1)) ;;
+          esac
+          ;;
+      esac
+    done < "$log_file"
+  }
+
+  if test_cross_connection_unattributable_errors_stay_loud > "$TMP/unattributable-mutation-output" 2>&1; then
+    fail "unattributable mutation accepted an unknown connection"
+  fi
+  grep -F 'FAIL: cross-unattributable recovered: expected 0, got 2' "$TMP/unattributable-mutation-output" > /dev/null \
+    || fail "unattributable mutation failed without the expected assertion"
+  printf 'Unattributable mutation: FAIL: cross-unattributable recovered: expected 0, got 2\n'
+}
+
+test_no_errors
+test_recovery_patterns_match_source
+test_attributed_recovered_connection_error
+test_source_derived_error_shapes_recover_by_connection
+test_cross_connection_unattributable_errors_stay_loud
+test_attributed_cross_connection_errors_stay_loud
+test_attributed_unrecovered_connection_error
+test_other_error_is_unexplained
+test_recovery_requirement_mutation_is_caught
+test_shared_counter_mutation_is_caught
+test_unattributable_quiet_mutation_is_caught
+printf 'PASS: redeploy log classifier\n'