fleetd #361 review round 2: pin isMissingQueue's false branch

The reviewer's mutation (isMissingQueue always returns true) restored the exact overstatement fleetd #361 exists to fix -- every declare failure reading as a confirmed absence -- and still left mvn clean install green (1389/1389), because no test drove a non-404 shape through inspect(). The false branch was the whole discriminator between MailboxState.absent() and MailboxState.unknown(), unpinned. Widened LeadMailbox.isMissingQueue from private to package-private and added LeadMailboxIsMissingQueueTest: five hermetic tests (no broker) covering the true case and all three false shapes isMissingQueue's own javadoc lists -- a different reply code, a ShutdownSignalException whose reason isn't a Channel.Close, and an IOException with no such cause at all (plus an IOException wrapping an unrelated exception type). Re-ran the reviewer's exact mutation locally: 4 of 5 new tests went red with the expected assertion messages; reverted, and mvn clean install is green again at 1394/1394 (1389 + 5 new). Also added a one-line javadoc note on LeadMailbox.inspect being honest about which of its two RuntimeException catches is proven by a test (the createChannel() one, end-to-end against a real broker) and which stays purely defensive (the declare-site one, for a connection-drops- mid-call race no test drives on purpose).
fleetd #361 review: fix false-negative absent, uncancelled probes, and a throw contract gap
2026-09-05 13:12:24 +07:00 · 2026-09-05 13:01:48 +07:00 · 2026-09-05 12:44:29 +07:00 · 2026-09-05 06:09:29 +07:00 · 2026-09-05 06:05:52 +07:00 · 2026-09-05 06:02:29 +07:00
20 changed files with 1420 additions and 66 deletions
@@ -821,10 +821,17 @@ guard:
 #              across every daemon sharing this vhost.
 #   prefetch → consumer basicQos, capping how many unacked messages the mailbox holds in-heap.
 #              Default 32 when omitted.
+#   peers    → fleetd #361: the coord-ids of the OTHER daemons on this vhost, declared by the
+#              operator (the daemon never guesses). fleet_list reports each one's live reachability
+#              (a passive queue check, never a presence protocol) alongside this daemon's own
+#              mailbox state. Omit, or leave empty, for a daemon with no known peers yet — an
+#              undeclared peer can still reach you and be reached by fleet_send, it just will not
+#              show up as a row in fleet_list.
 # coordinator:
 #   uriEnv: LEAD_COORD_URI
 #   selfId: mac-opus
 #   prefetch: 32
+#   peers: [fleet01-lead]

 # Active push-to-primary (CB-307 Stage 3). When a worker reply lands with no open fleet_send,
 # the ReplyPushLoop injects a *drain nudge* (never the payload) into the primary's own herdr
@@ -652,7 +652,12 @@ public final class Fleetd {
                quarantineSource,
                leadMailbox,
                outageSource,
-                new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)));
+                new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)),
+                // fleetd #361: the operator-declared peers this daemon's fleet_list should try to
+                // reach. Read from the SAME snapshot leadMailbox itself opened from (cfg.coordinator()),
+                // not the live config.get() — coordinator wiring is already boot-time-fixed (see
+                // leadMailbox above), so peers follows the same rule rather than half hot-reloading.
+                cfg.coordinator() == null ? List.of() : cfg.coordinator().peers());

        // CB-637: the receive half. Only constructed when a lead mailbox actually opened — with no
        // coordinator (or an unreachable one) there is nothing to deliver, so no scheduler is
@@ -888,12 +888,18 @@ public record FleetConfig(
     *                 {@code null} ⇒ kept as {@code null} (no self id configured).
     * @param prefetch the consumer's {@code basicQos} prefetch count. {@code null}/non-positive ⇒
     *                 {@link LeadMailbox#DEFAULT_PREFETCH}.
+     * @param peers    fleetd #361: the coord-ids the operator declares as this daemon's peers — the
+     *                 daemon never guesses who else exists. {@code fleet_list} reports each one's
+     *                 live reachability. Blank entries are dropped; {@code null} ⇒ an empty list, so
+     *                 a config written before this field existed still parses unchanged.
     */
    @JsonIgnoreProperties(ignoreUnknown = true)
-    public record Coordinator(String uri, String uriEnv, String selfId, Integer prefetch) {
+    public record Coordinator(String uri, String uriEnv, String selfId, Integer prefetch, List<String> peers) {

        public Coordinator {
            selfId = (selfId == null || selfId.isBlank()) ? null : selfId;
+            peers = peers == null ? List.of()
+                    : peers.stream().filter(p -> p != null && !p.isBlank()).toList();
        }

        /** True when a {@code uriEnv} is configured by name, whether or not its variable resolves. */
@@ -44,6 +44,7 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
@@ -101,6 +102,8 @@ public final class FleetMcp {
    private final LeadSeatSource leadSeats;
    /** CB-637: this daemon's lead-to-lead channel; {@code null} when no coordinator is configured. */
    private final LeadChannel leadChannel;
+    /** fleetd #361: {@code coordinator.peers} — see {@link CoordinationSource}. Empty when unset. */
+    private final List<String> peers;

    /** Capacity facts used by {@code fleet_list}; production must supply the placement live count. */
    public record CapacitySource(Function<String, Integer> liveCount, Function<String, Integer> maxLoad,
@@ -164,6 +167,32 @@ public final class FleetMcp {
        public static LeadSeatSource none() { return new LeadSeatSource(_ -> 0); }
    }

+    /**
+     * fleetd #361: peer-visibility facts for {@code fleet_list}'s {@code coordinator} row — this
+     * daemon's own {@link LeadChannel} (for its self mailbox state and held messages) plus the
+     * coord-ids the operator has declared as peers ({@code coordinator.peers}). Bundled as its own
+     * Source, the same idiom as {@link OutageSource}/{@link QuarantineSource}/{@link LeadSeatSource},
+     * so {@code listFleet}'s already-long overload chain gains exactly one new required parameter
+     * instead of a further bare positional argument.
+     *
+     * <p>Every mailbox look this triggers goes through {@link LeadChannel#inspect}, which is
+     * specified to run on its own disposable channel — never the channel {@link LeadChannel#publish}
+     * or the consume loop depends on — so a peer that happens to be down, or the coordination broker
+     * itself being unreachable, can never take {@code fleet_send}/{@code LeadCoordLoop}'s own path
+     * down with it. See {@code FleetMcp.probe} for the additional timeout bound on top of that.
+     *
+     * @param leadChannel this daemon's own channel, or {@code null} when no coordinator is configured
+     * @param peers       the coord-ids declared under {@code coordinator.peers}, or empty
+     */
+    public record CoordinationSource(LeadChannel leadChannel, List<String> peers) {
+        public CoordinationSource {
+            peers = peers == null ? List.of() : List.copyOf(peers);
+        }
+
+        /** Inert source — no coordinator row is ever reported. */
+        public static CoordinationSource none() { return new CoordinationSource(null, List.of()); }
+    }
+
    /**
     * @param callers resolves each call's {@link Principal}; {@code null} disables authorization.
     *                This surface needs its own enforcement: {@code /mcp} is a raw servlet on
@@ -211,19 +240,34 @@ public final class FleetMcp {
    }

    /**
-     * As above, with fleetd #176 lead-seat facts (see {@link LeadSeatSource}). This is what
-     * {@code Fleetd.main} actually wires up.
-     *
-     * @param leadSeats required — pass {@link LeadSeatSource#none()} for a caller that does not want
-     *                  the feature, never a defaulting overload (the same rule {@code quarantine} and
-     *                  {@code outage} follow).
+     * As above, with fleetd #176 lead-seat facts (see {@link LeadSeatSource}).
     */
    public FleetMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
                     ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
                      CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage,
                      QuarantineSource quarantine, LeadChannel leadChannel, OutageSource outage,
                      LeadSeatSource leadSeats) {
+        this(messages, workers, sessions, identity, presence, primaryRegistry, callers, metrics, capacity,
+                healthCoverage, quarantine, leadChannel, outage, leadSeats, List.of());
+    }
+
+    /**
+     * As above, with fleetd #361 {@code coordinator.peers} (see {@link CoordinationSource}). This is
+     * what {@code Fleetd.main} actually wires up.
+     *
+     * @param leadSeats required — pass {@link LeadSeatSource#none()} for a caller that does not want
+     *                  the feature, never a defaulting overload (the same rule {@code quarantine} and
+     *                  {@code outage} follow).
+     * @param peers     the coord-ids declared under {@code coordinator.peers}; empty when unset or
+     *                  when {@code leadChannel} is {@code null}.
+     */
+    public FleetMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
+                     ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
+                      CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage,
+                      QuarantineSource quarantine, LeadChannel leadChannel, OutageSource outage,
+                      LeadSeatSource leadSeats, List<String> peers) {
        this.leadChannel = leadChannel;
+        this.peers = peers == null ? List.of() : List.copyOf(peers);
        this.capacity = capacity;
        this.quarantine = Objects.requireNonNull(quarantine, "quarantine");
        this.outage = Objects.requireNonNull(outage, "outage");
@@ -361,7 +405,7 @@ public final class FleetMcp {
                    return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, outage,
                            leadSeats, callers == null ? Map.of() : callers.leads(),
                            callerTerminal(exchange),
-                            leadChannel == null ? null : leadChannel.selfCoordId());
+                            new CoordinationSource(leadChannel, peers));
                };
        BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> stopHandler =
                (exchange, req) -> {
@@ -697,6 +741,15 @@ public final class FleetMcp {
     * turned into a tool error naming the coord-id. It is never allowed to escape as a crash: an
     * unreachable peer is an ordinary outcome of addressing a fleet you do not control.
     *
+     * <p><strong>fleetd #361: the success text is honest about what "durably confirmed" does and
+     * does not mean.</strong> The broker's publisher confirm proves the message is durably queued —
+     * it says nothing about whether the peer's pane has, or ever will, receive it. After a
+     * successful publish this looks at the target mailbox's consumer count (via
+     * {@link LeadChannel#inspect}, bounded and never allowed to fail the call — see {@link #probe})
+     * and appends a warning when it is zero: that is the observable form of "nobody is reading this
+     * right now". A zero-consumer publish is still reported as a SUCCESS, never an error — the
+     * message is safely queued and will be read once a daemon owning that coord-id connects.
+     *
     * @param leadChannel this daemon's channel, or {@code null} when no coordinator is configured
     */
    static McpSchema.CallToolResult sendToLead(LeadChannel leadChannel, String coordId, String content,
@@ -724,7 +777,16 @@ public final class FleetMcp {
                    + ". Check that a daemon is running with coordinator.selfId=\"" + coordId
                    + "\" and is connected to the same coordination broker.");
        }
-        return text("delivered to peer lead " + coordId + " (msgId " + msg.msgId() + ")");
+        String result = "published to peer lead \"" + coordId + "\"'s mailbox and durably confirmed "
+                + "by the broker (msgId " + msg.msgId() + ").";
+        LeadChannel.MailboxState state = probe(leadChannel, coordId);
+        if (state.exists() && state.consumers() == 0) {
+            result += " Warning: that mailbox currently has NO consumers attached — nobody is reading "
+                    + "it right now. The message is safely queued and will be delivered once a daemon "
+                    + "with coordinator.selfId=\"" + coordId + "\" is running and connected; until then "
+                    + "it will not reach that lead's pane.";
+        }
+        return text(result);
    }

    /**
@@ -1110,7 +1172,8 @@ public final class FleetMcp {
    static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                               CapacitySource capacity, HealthCoverageSource healthCoverage,
                                               QuarantineSource quarantine, Map<String, String> leads, String selfTerm) {
-        return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, leads, selfTerm, null);
+        return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, leads, selfTerm,
+                CoordinationSource.none());
    }

    /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1119,33 +1182,33 @@ public final class FleetMcp {
                                               QuarantineSource quarantine, OutageSource outage,
                                               Map<String, String> leads, String selfTerm) {
        return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, outage,
-                LeadSeatSource.none(), leads, selfTerm, null);
+                LeadSeatSource.none(), leads, selfTerm, CoordinationSource.none());
    }

    /**
-     * As above, additionally reporting this daemon's own lead coordination id (CB-637) when one is
-     * configured and its channel opened. There is no peer-discovery surface yet — a lead addresses a
-     * peer by a coord-id it was told — so this row exists to answer the one question the operator
-     * cannot answer any other way: what is MY coord-id, the one a peer must use to reach me. It is
-     * omitted entirely when no coordinator is configured, so an ordinary fleet's output is unchanged.
+     * As above, additionally reporting this daemon's own lead coordination state (CB-637, fleetd
+     * #361) when a coordinator is configured and its channel opened — see {@link #coordinatorView}
+     * for the shape. Omitted entirely when no coordinator is configured, so an ordinary fleet's
+     * output is unchanged.
     *
-     * @param selfCoordId this daemon's coord-id, or {@code null} when lead coordination is off
+     * @param coordination this daemon's lead channel plus its declared peers, or
+     *                      {@link CoordinationSource#none()} when lead coordination is off
     */
    static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                               CapacitySource capacity, HealthCoverageSource healthCoverage,
                                               QuarantineSource quarantine, Map<String, String> leads, String selfTerm,
-                                               String selfCoordId) {
+                                               CoordinationSource coordination) {
        return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, OutageSource.none(),
-                LeadSeatSource.none(), leads, selfTerm, selfCoordId);
+                LeadSeatSource.none(), leads, selfTerm, coordination);
    }

    /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
    static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                               CapacitySource capacity, HealthCoverageSource healthCoverage,
                                               QuarantineSource quarantine, OutageSource outage,
-                                               Map<String, String> leads, String selfTerm, String selfCoordId) {
+                                               Map<String, String> leads, String selfTerm, CoordinationSource coordination) {
        return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, outage,
-                LeadSeatSource.none(), leads, selfTerm, selfCoordId);
+                LeadSeatSource.none(), leads, selfTerm, coordination);
    }

    /** As above, plus fleetd #176 lead-seat facts (see {@link LeadSeatSource}). */
@@ -1153,7 +1216,7 @@ public final class FleetMcp {
                                               CapacitySource capacity, HealthCoverageSource healthCoverage,
                                               QuarantineSource quarantine, OutageSource outage,
                                               LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
-                                               String selfCoordId) {
+                                               CoordinationSource coordination) {
        try {
            Map<String, Agent> live = workers.list().stream()
                    .map(Agent.class::cast)
@@ -1175,8 +1238,9 @@ public final class FleetMcp {
            Map<String, Object> result = new LinkedHashMap<>();
            result.put("leads", leadRows); result.put("members", out);
            result.put("healthCoverage", healthCoverage.value().get());
-            if (selfCoordId != null && !selfCoordId.isBlank()) {
-                result.put("coordinator", Map.of("selfId", selfCoordId, "configured", true));
+            Map<String, Object> coordinatorRow = coordinatorView(coordination);
+            if (coordinatorRow != null) {
+                result.put("coordinator", coordinatorRow);
            }
            if (capacity.available()) result.put("capacity", profiles.stream()
                    .map(profile -> capacityView(profile, capacity.liveCount(), capacity.maxLoad(), roster, messages,
@@ -1187,6 +1251,140 @@ public final class FleetMcp {
        }
    }

+    /**
+     * fleetd #361: the {@code coordinator} row — this daemon's own coord-id and mailbox state, the
+     * messages currently held for it, and the live reachability of every operator-declared peer.
+     * {@code null} (the row is then omitted entirely) when lead coordination is off, so an ordinary
+     * fleet's {@code fleet_list} output is byte-identical to before this feature existed.
+     *
+     * <p>Every peer/self mailbox look goes through {@link #probe}, which bounds each
+     * {@link LeadChannel#inspect} call to {@link #PEER_PROBE_TIMEOUT_MS} and never lets it throw —
+     * a coordination broker that is down or slow degrades this row toward "unreachable"/"unknown"
+     * counts, it can never make {@code fleet_list} itself slow or fail. {@code held} comes from
+     * {@link LeadChannel#peek}, a pure in-memory read with no broker round trip, so it is never
+     * subject to that bound.
+     */
+    private static Map<String, Object> coordinatorView(CoordinationSource coordination) {
+        LeadChannel channel = coordination.leadChannel();
+        if (channel == null) {
+            return null;
+        }
+        String selfId = channel.selfCoordId();
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("selfId", selfId);
+        row.put("configured", true);
+        row.put("mailbox", mailboxView(probe(channel, selfId)));
+        row.put("held", channel.peek().stream().map(FleetMcp::heldView).toList());
+        row.put("peers", coordination.peers().stream().map(p -> peerView(channel, p)).toList());
+        return row;
+    }
+
+    /**
+     * fleetd #361: render a {@link LeadChannel.MailboxState} without ever presenting an unmeasured
+     * fact as a measured one. {@code status} is the tri-state itself — {@code "exists"},
+     * {@code "absent"} (the broker positively confirmed no such queue), or {@code "unknown"} (the
+     * probe could not determine either way: down, unreachable, or timed out). {@code pending}/
+     * {@code consumers} are included ONLY when {@code status == "exists"} — a reader must never see
+     * them default to {@code 0} for a mailbox this call never actually measured. This is the fix for
+     * the review finding that a collapsed {@code absent()} rendered a self-probe timeout as
+     * "pending: 0, consumers: 0", indistinguishable from an actually-empty, actually-unread mailbox.
+     */
+    private static Map<String, Object> mailboxView(LeadChannel.MailboxState state) {
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("status", state.exists() ? "exists" : state.known() ? "absent" : "unknown");
+        if (state.exists()) {
+            row.put("pending", state.pending());
+            row.put("consumers", state.consumers());
+        }
+        return row;
+    }
+
+    /** One held-for-me message: enough to identify it and see roughly what it says, never the whole body. */
+    private static Map<String, Object> heldView(LeadMessage m) {
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("msgId", m.msgId());
+        row.put("from", m.from());
+        row.put("preview", preview(m.content()));
+        return row;
+    }
+
+    /** Cap a held message's content to a short preview — {@code fleet_list} must never dump a full body. */
+    private static final int HELD_PREVIEW_MAX_CHARS = 80;
+
+    private static String preview(String content) {
+        if (content == null) {
+            return "";
+        }
+        return content.length() <= HELD_PREVIEW_MAX_CHARS
+                ? content
+                : content.substring(0, HELD_PREVIEW_MAX_CHARS) + "…";
+    }
+
+    /**
+     * One declared peer's row: its coord-id, then the same tri-state {@link #mailboxView} shape.
+     * Deliberately no boolean "reachable" field — that collapsed "confirmed gone" and "could not
+     * check" into the same {@code false}, which is exactly the review finding this row now avoids:
+     * an operator reading {@code status} can tell "fleet01 is down" (a {@code coordinator.selfId}
+     * nobody has ever run) apart from "my own broker is slow or unreachable right now".
+     */
+    private static Map<String, Object> peerView(LeadChannel channel, String coordId) {
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("coordId", coordId);
+        row.putAll(mailboxView(probe(channel, coordId)));
+        return row;
+    }
+
+    /**
+     * fleetd #361: how long {@code fleet_list} waits on any single {@link LeadChannel#inspect} call
+     * before giving up on it — see {@link #probe}.
+     */
+    private static final long PEER_PROBE_TIMEOUT_MS = 1_500L;
+
+    /**
+     * Dedicated pool for {@link LeadChannel#inspect} calls so a slow one blocks only its own virtual
+     * thread, never the MCP request thread calling {@code fleet_list}. Not a <em>bounded</em> pool —
+     * {@code newThreadPerTaskExecutor} starts a fresh virtual thread per call with no cap on how many
+     * run at once; virtual threads make that cheap, not bounded. What actually keeps a hung probe
+     * from accumulating forever is the {@link Future#cancel} in {@link #probe}, not a pool limit.
+     */
+    private static final java.util.concurrent.ExecutorService PEER_PROBE_POOL =
+            java.util.concurrent.Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("fleet-peer-probe-", 0).factory());
+
+    /**
+     * fleetd #361: {@link LeadChannel#inspect}, bounded to {@link #PEER_PROBE_TIMEOUT_MS} and never
+     * allowed to throw or hang the caller — a coordination broker that is unreachable or slow
+     * degrades to {@link LeadChannel.MailboxState#unknown} (never {@code absent}: a timeout proves
+     * nothing about whether the mailbox exists) rather than making {@code fleet_list} slow or
+     * failing it. {@code inspect} itself is already specified to never throw, but this is the seam
+     * that also survives an implementation that does, or one that blocks indefinitely on a dead
+     * connection.
+     *
+     * <p><strong>A timeout cancels the orphaned task</strong> rather than abandoning it. Before this,
+     * {@code get(timeout)} on a hung {@code inspect} left the submitted task running forever on its
+     * own virtual thread, holding the AMQP channel it had already opened — against a broker that
+     * hangs rather than fails fast, every {@code fleet_list} call would orphan one more channel until
+     * the connection's channel-max (2047 by default) was exhausted, which would break {@link
+     * LeadChannel#publish} too. {@link Future#cancel(boolean) cancel(true)} interrupts the orphaned
+     * task's thread; {@link LeadMailbox#inspect} has no interruptible wait of its own to catch that,
+     * but the underlying AMQP RPC continuation does block on one, so the interrupt reaches it and the
+     * task's {@code finally} still closes the probe channel it opened rather than leaking it forever.
+     */
+    private static LeadChannel.MailboxState probe(LeadChannel channel, String coordId) {
+        return probe(channel, coordId, PEER_PROBE_TIMEOUT_MS);
+    }
+
+    /** As {@link #probe(LeadChannel, String)}, with an explicit timeout — a seam for tests. */
+    static LeadChannel.MailboxState probe(LeadChannel channel, String coordId, long timeoutMs) {
+        java.util.concurrent.Future<LeadChannel.MailboxState> future =
+                PEER_PROBE_POOL.submit(() -> channel.inspect(coordId));
+        try {
+            return future.get(timeoutMs, TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            future.cancel(true); // best-effort: don't leave a hung probe (and its channel) running forever
+            return LeadChannel.MailboxState.unknown(coordId);
+        }
+    }
+
    /**
     * Capacity is advisory only. {@code reclaimable} says there is no bridge work, not that fleetd
     * may stop the member: the bridge has capacity facts but no work list, and choosing work needs
@@ -1377,8 +1575,12 @@ public final class FleetMcp {
                                + "routes your answer back into the same turn (omit for a normal delegation)"),
                        "coordId", stringProp("A peer LEAD's coordination id — delivers content to that "
                                + "lead's durable mailbox on the shared coordination broker, which works "
-                                + "across hosts. Mutually exclusive with sessionId and turnId. Your own "
-                                + "coordId is reported by fleet_list.")),
+                                + "across hosts. Mutually exclusive with sessionId and turnId. Success means "
+                                + "the message is durably queued and confirmed by the broker, with a warning "
+                                + "if that mailbox has no consumers attached right now (queued, but nobody is "
+                                + "reading it yet) — it does not mean the peer's pane has seen it. Your own "
+                                + "coordId, this daemon's mailbox state, and every coordinator.peers entry's "
+                                + "live reachability are reported by fleet_list's coordinator row.")),
                        List.of("content")));
    }

@@ -1486,7 +1688,13 @@ public final class FleetMcp {
                        + "for a quarantined profile's credential (see fleet_profiles), whatever its "
                        + "maxLoad/live — with credentialId and quarantinedForSeconds naming the "
                        + "quarantine, so 'free: 0, busy' can be told apart from 'free: 0, refusing "
-                        + "for N seconds'.",
+                        + "for N seconds'. When lead-to-lead coordination is configured, a 'coordinator' "
+                        + "object reports this daemon's own coord-id ('selfId') and mailbox state "
+                        + "('mailbox': pending/consumers), the messages currently held for it ('held': "
+                        + "msgId/from/preview, never the full body), and one row per coordinator.peers "
+                        + "coord-id ('peers': coordId/reachable, plus pending/consumers when reachable) — "
+                        + "this is peer DISCOVERY for cross-host leads, distinct from the local 'leads' "
+                        + "array above. It is omitted entirely when no coordinator is configured.",
                objectSchema(Map.of(), List.of()));
    }

@@ -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());
+    }
+}
@@ -4,8 +4,8 @@ import java.util.List;

 /**
 * The lead-to-lead message channel this daemon speaks, as its callers need it — one lead's own
- * mailbox: publish to a peer's coord-id, look at what has arrived for me, and ack what I have
- * delivered.
+ * mailbox: publish to a peer's coord-id, look at what has arrived for me, ack what I have
+ * delivered, and (fleetd #361) inspect any coord-id's mailbox from the outside without owning it.
 *
 * <p>Extracted from {@link LeadMailbox} purely as a seam. {@code LeadMailbox} is the one production
 * implementation and owns a live AMQP connection, so a test that wanted to exercise the routing in
@@ -37,4 +37,78 @@ public interface LeadChannel {

    /** This daemon's own lead coordination id — the mailbox it owns, and the {@code from} it sends as. */
    String selfCoordId();
+
+    /**
+     * A non-destructive look at {@code coordId}'s mailbox — does it exist, how many messages are
+     * waiting on it, and how many consumers are attached — without owning, consuming, or otherwise
+     * changing it. {@code consumers == 0} on an existing mailbox is the observable form of "nobody
+     * is reading this right now": a publish to it will sit queued rather than reach a pane.
+     *
+     * <p><strong>Never throws</strong> — this is a best-effort fact-finding call, not an operation a
+     * caller must handle failing. But it must never turn "I could not check" into a false negative:
+     * {@link MailboxState#absent(String)} means the broker positively confirmed there is no such
+     * queue, and {@link MailboxState#unknown(String)} — a distinct value — means the look could not
+     * be completed at all (broker unreachable, timed out, connection closed). A caller that
+     * collapses those two into one, as fleetd #361 initially did, cannot tell "that peer is down"
+     * from "I could not check", and a reader of {@code pending}/{@code consumers} cannot tell a
+     * measured zero from a zero standing in for "not measured".
+     *
+     * <p><strong>Must never share fate with {@link #publish} or {@link #peek}/{@link #ack}.</strong>
+     * fleetd #361: in AMQP 0-9-1 a passive queue declare of a queue that does not exist closes the
+     * channel it was declared on with a 404. An implementation backed by a real broker connection
+     * must inspect on a channel it can afford to lose — never the channel {@link #publish} or the
+     * consume loop depends on — so that looking at a peer that happens to be down can never break
+     * this daemon's own send or receive path.
+     */
+    MailboxState inspect(String coordId);
+
+    /**
+     * The result of {@link #inspect}. {@code presence} tells apart three states a caller must not
+     * conflate: a confirmed-existing mailbox ({@link Presence#EXISTS}, the only case where
+     * {@code pending}/{@code consumers} are measured facts), a confirmed-absent one
+     * ({@link Presence#ABSENT} — the broker positively said "no such queue"), and one this call
+     * simply could not determine ({@link Presence#UNKNOWN} — broker unreachable, timed out,
+     * connection closed). {@code pending}/{@code consumers} are always {@code 0} and meaningless
+     * outside {@link Presence#EXISTS}; a renderer must gate on {@link #exists()} (or {@code
+     * presence} directly), never present them as measured otherwise.
+     *
+     * @param coordId   the coord-id inspected
+     * @param presence  whether the mailbox is confirmed to exist, confirmed absent, or unknown
+     * @param pending   messages ready for delivery but not yet in a consumer's hands (0 unless EXISTS)
+     * @param consumers how many consumers are attached (0 unless EXISTS)
+     */
+    record MailboxState(String coordId, Presence presence, int pending, int consumers) {
+
+        /** Whether {@link #inspect} was able to reach a definite answer, of either kind. */
+        public enum Presence { EXISTS, ABSENT, UNKNOWN }
+
+        /** {@code true} only when the broker confirmed this exact queue is currently declared. */
+        public boolean exists() {
+            return presence == Presence.EXISTS;
+        }
+
+        /**
+         * {@code true} when {@link #inspect} reached a definite answer (exists or confirmed
+         * absent); {@code false} when it could not determine either way. A caller must never treat
+         * {@code !known()} the same as a confirmed absence — the mailbox may well exist.
+         */
+        public boolean known() {
+            return presence != Presence.UNKNOWN;
+        }
+
+        /** The broker confirmed this queue exists, with these measured counts. */
+        public static MailboxState exists(String coordId, int pending, int consumers) {
+            return new MailboxState(coordId, Presence.EXISTS, pending, consumers);
+        }
+
+        /** The broker positively confirmed there is no such queue (e.g. a 404 on passive declare). */
+        public static MailboxState absent(String coordId) {
+            return new MailboxState(coordId, Presence.ABSENT, 0, 0);
+        }
+
+        /** The look could not be completed — broker unreachable, timed out, or connection closed. */
+        public static MailboxState unknown(String coordId) {
+            return new MailboxState(coordId, Presence.UNKNOWN, 0, 0);
+        }
+    }
 }
@@ -10,6 +10,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.ShutdownSignalException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

@@ -122,17 +123,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);
@@ -254,6 +260,89 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
        }
    }

+    /**
+     * fleetd #361: look at {@code coordId}'s mailbox on a fresh, immediately-closed throwaway
+     * channel — never {@link #channel} (consume/ack) or {@link #publishChannel} (publish). A
+     * passive queue declare of a queue that does not exist closes the channel it was declared on
+     * with a 404; using a disposable probe channel means that closure can never touch either
+     * long-lived channel this instance depends on for {@link #publish} or the consume loop.
+     *
+     * <p>Classifies failures rather than collapsing them, both measured against a real broker in
+     * {@code LeadMailboxTest} rather than assumed from the AMQP 0-9-1 spec text:
+     * <ul>
+     *   <li>a genuine 404 — an {@link IOException} wrapping a {@link ShutdownSignalException} whose
+     *       {@link AMQP.Channel.Close#getReplyCode()} is {@code 404} — reports
+     *       {@link MailboxState#absent}; every other declare failure reports
+     *       {@link MailboxState#unknown} instead of quietly becoming the same "absent" value;
+     *   <li>{@code catch (RuntimeException e)} on both attempts matters as much as the checked
+     *       catches: a connection that is already closed makes {@link Connection#createChannel()}
+     *       throw {@link com.rabbitmq.client.AlreadyClosedException} (a {@link RuntimeException},
+     *       not an {@link IOException}) — an {@code inspect} that only caught {@code IOException}
+     *       would let that escape, breaking the "never throws" contract this method promises.
+     * </ul>
+     *
+     * <p><strong>Honesty about which catch is measured and which is defensive:</strong> the
+     * {@code createChannel()} catch above is exercised end-to-end against a real broker by
+     * {@code LeadMailboxTest.inspectReportsUnknownRatherThanThrowingWhenTheConnectionIsAlreadyClosed}.
+     * The second {@code catch (RuntimeException e)}, around the passive declare itself — for the
+     * narrower race where the connection drops <em>between</em> {@code createChannel()} succeeding
+     * and the declare landing — has no such test; reaching it needs a connection that dies at that
+     * exact instant, which is not a scenario this suite drives on purpose. It stays purely
+     * defensive: correct by the same reasoning as the first catch, but unproven the way the first
+     * one is proven.
+     */
+    @Override
+    public MailboxState inspect(String coordId) {
+        String queue = queueName(coordId);
+        Channel probe;
+        try {
+            probe = connection.createChannel();
+        } catch (IOException | RuntimeException e) {
+            log.debug("lead mailbox inspect: cannot open a probe channel for {}: {}", coordId, e.toString());
+            return MailboxState.unknown(coordId);
+        }
+        try {
+            AMQP.Queue.DeclareOk declared = probe.queueDeclarePassive(queue);
+            return MailboxState.exists(coordId, declared.getMessageCount(), declared.getConsumerCount());
+        } catch (IOException e) {
+            // The broker (or the client library) has already closed `probe` for us either way; only
+            // a confirmed 404 means "no such queue" — anything else (a different declare failure) is
+            // "could not determine", never silently reported as the same value as a genuine absence.
+            return isMissingQueue(e) ? MailboxState.absent(coordId) : MailboxState.unknown(coordId);
+        } catch (RuntimeException e) {
+            // E.g. the connection dropped between createChannel() and the declare landing.
+            log.debug("lead mailbox inspect: declare failed unexpectedly for {}: {}", coordId, e.toString());
+            return MailboxState.unknown(coordId);
+        } finally {
+            try {
+                if (probe.isOpen()) {
+                    probe.close();
+                }
+            } catch (Exception e) {
+                log.debug("lead mailbox inspect: probe channel close for {}: {}", coordId, e.toString());
+            }
+        }
+    }
+
+    /**
+     * {@code true} only for the specific shape a missing-queue passive declare actually produces —
+     * measured against a real broker, not assumed from the spec text (see {@code
+     * LeadMailboxTest.passiveDeclareOfAMissingQueueThrowsAnIOExceptionWrappingA404ShutdownSignal}):
+     * an {@link IOException} whose cause is a {@link ShutdownSignalException} carrying an
+     * {@link AMQP.Channel.Close} reason with {@code replyCode == 404}. Any other shape (a different
+     * reply code, a {@code ShutdownSignalException} cause whose reason is not a
+     * {@code Channel.Close}, or no cause at all) is a declare failure of some other kind and must
+     * not be read as "confirmed absent" — pinned hermetically, with no broker needed, by
+     * {@code LeadMailboxIsMissingQueueTest} for exactly those three false shapes. Package-private
+     * (not {@code private}) so that test can call it directly.
+     */
+    static boolean isMissingQueue(IOException e) {
+        if (!(e.getCause() instanceof ShutdownSignalException sse)) {
+            return false;
+        }
+        return sse.getReason() instanceof AMQP.Channel.Close close && close.getReplyCode() == AMQP.NOT_FOUND;
+    }
+
    /** Convenience: {@link #peek} the current snapshot, then {@link #ack} every message in it. */
    public List<LeadMessage> drain() {
        List<LeadMessage> snapshot = peek();
@@ -80,7 +80,7 @@ class FleetdLeadMailboxSelectionTest {
    @Test
    void opensTheMailboxWhenAUriAndSelfIdAreConfigured() {
        var opener = new RecordingOpener();
-        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null);
+        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null, null);

        Fleetd.openLeadMailbox(coordinator, Map.of(), opener);

@@ -94,7 +94,7 @@ class FleetdLeadMailboxSelectionTest {
    void honoursUriEnvOverALiteralUri() {
        var opener = new RecordingOpener();
        var coordinator = new FleetConfig.Coordinator("amqp://stale:stale@old:5672/x", "COORD_URI",
-                "mac-opus", 8);
+                "mac-opus", 8, null);

        Fleetd.openLeadMailbox(coordinator, Map.of("COORD_URI", RESOLVED_URI), opener);

@@ -105,7 +105,7 @@ class FleetdLeadMailboxSelectionTest {
    @Test
    void turnsOffWhenUriEnvDoesNotResolve() {
        var opener = new RecordingOpener();
-        var coordinator = new FleetConfig.Coordinator(null, "COORD_URI", "mac-opus", null);
+        var coordinator = new FleetConfig.Coordinator(null, "COORD_URI", "mac-opus", null, null);

        assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener));

@@ -116,7 +116,7 @@ class FleetdLeadMailboxSelectionTest {
    void warnsAndStaysOffWhenSelfIdIsMissing() {
        var appender = captureFleetdLogs();
        var opener = new RecordingOpener();
-        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, null, null);
+        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, null, null, null);

        assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener));

@@ -131,7 +131,7 @@ class FleetdLeadMailboxSelectionTest {
        var appender = captureFleetdLogs();
        var opener = new RecordingOpener();
        opener.unreachable = true;
-        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null);
+        var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null, null);

        assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener),
                "a down coordination broker turns the feature off; it must never take the daemon down");
@@ -104,7 +104,7 @@ class ConfigRefTopLevelReportingCoverageTest {
        v.put("configReload", new FleetConfig.ConfigReload(true, 10));
        v.put("quarantineCooldownSeconds", 1800);
        v.put("memberCredentials", null);
-        v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-a", null, "self-a", 1));
+        v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-a", null, "self-a", 1, null));
        v.put("worktreeGroup", "group-a");
        v.put("memberLoginShell", null);
        assertNamesMatchComponents(v);
@@ -144,7 +144,7 @@ class ConfigRefTopLevelReportingCoverageTest {
        v.put("configReload", new FleetConfig.ConfigReload(false, 20));
        v.put("quarantineCooldownSeconds", 3600);
        v.put("memberCredentials", null);
-        v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-b", null, "self-b", 2));
+        v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-b", null, "self-b", 2, null));
        v.put("worktreeGroup", "group-b");
        v.put("memberLoginShell", null);
        assertNamesMatchComponents(v);
@@ -1189,7 +1189,7 @@ class FleetConfigTest {
    @Test
    void coordinatorEffectiveUriHonorsUriEnv() {
        FleetConfig.Coordinator withEnv = new FleetConfig.Coordinator(
-                "amqp://stale-clear-text@127.0.0.1:5672/coord", "LEAD_COORD_URI", "fleet01-lead", null);
+                "amqp://stale-clear-text@127.0.0.1:5672/coord", "LEAD_COORD_URI", "fleet01-lead", null, null);

        assertEquals("amqp://from-env@127.0.0.1:5672/coord",
                withEnv.effectiveUri(Map.of("LEAD_COORD_URI", "amqp://from-env@127.0.0.1:5672/coord")),
@@ -1200,12 +1200,61 @@ class FleetConfigTest {
                "a blank uriEnv variable must not fall back to the literal uri");

        FleetConfig.Coordinator noEnv = new FleetConfig.Coordinator(
-                "amqp://guest:guest@127.0.0.1:5672/coord", null, null, null);
+                "amqp://guest:guest@127.0.0.1:5672/coord", null, null, null, null);
        assertEquals("amqp://guest:guest@127.0.0.1:5672/coord", noEnv.effectiveUri(Map.of()),
                "the literal uri is used when no uriEnv is configured");
        assertEquals(LeadMailbox.DEFAULT_PREFETCH, noEnv.prefetchOrDefault());
    }

+    /**
+     * fleetd #361: the live block ships as just {@code uriEnv} + {@code selfId} (see
+     * {@code Fleetd.example.yaml} / the operator's real {@code fleetd.yaml}, gitignored). That exact
+     * shape, with no {@code peers:} key at all, must keep parsing unchanged after this field is added.
+     */
+    @Test
+    void coordinatorBlockWithNoPeersKeyStillParses(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("coordinator-no-peers.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                coordinator:
+                  uriEnv: COORD_AMQP_URI
+                  selfId: mac
+                """);
+
+        FleetConfig cfg = FleetConfig.load(f);
+        assertNotNull(cfg.coordinator());
+        assertEquals("mac", cfg.coordinator().selfId());
+        assertEquals(List.of(), cfg.coordinator().peers(), "no peers: key means no configured peers, never null");
+    }
+
+    @Test
+    void coordinatorPeersParsesAndDropsBlankEntries(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("coordinator-peers.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                coordinator:
+                  selfId: mac
+                  uri: amqp://guest:guest@127.0.0.1:5672/coord
+                  peers:
+                    - fleet01
+                    - ""
+                    - fleet02
+                """);
+
+        FleetConfig cfg = FleetConfig.load(f);
+        assertEquals(List.of("fleet01", "fleet02"), cfg.coordinator().peers(),
+                "a blank peer entry must be dropped, never kept as an empty coord-id");
+    }
+
+    @Test
+    void coordinatorPeersDefaultsToEmptyWhenConstructedWithNull() {
+        FleetConfig.Coordinator c = new FleetConfig.Coordinator(
+                "amqp://guest:guest@127.0.0.1:5672/coord", null, "mac", null, null);
+        assertEquals(List.of(), c.peers(), "a null peers list must default to empty, never NPE downstream");
+    }
+
    @Test
    void absentWorktreeGroupLeavesItNull(@TempDir Path dir) throws Exception {
        Path f = dir.resolve("no-worktree-group.yaml");
@@ -92,7 +92,7 @@ class FleetConfigWithDefaultsPreservesEveryComponentTest {
        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("coordinator", new FleetConfig.Coordinator("amqp://coord-guard", null, "self-guard", 3, null));
        v.put("worktreeGroup", "group-guard");
        v.put("memberLoginShell", "/bin/zsh");
        assertNamesMatchComponents(v);
@@ -1,6 +1,7 @@
 package dev.ltms.fleet.mcp;

 import dev.ltms.fleet.msg.FakeLeadChannel;
+import dev.ltms.fleet.msg.LeadChannel;
 import dev.ltms.fleet.msg.LeadMessage;
 import io.modelcontextprotocol.spec.McpSchema;
 import org.junit.jupiter.api.Test;
@@ -93,4 +94,56 @@ class FleetMcpLeadCoordTest {
        assertTrue(FleetMcp.sendToLead(channel, PEER, "   ", null, null).isError());
        assertEquals(0, channel.published().size());
    }
+
+    /**
+     * fleetd #361: "delivered" overstated what publish actually proves — the broker's confirm means
+     * durably queued, not read. A zero-consumer target is the observable form of "this will not
+     * reach a pane right now", so the (still-successful) result must say so.
+     */
+    @Test
+    void warnsWhenThePeerMailboxHasNoConsumersButStillReportsSuccess() {
+        var channel = new FakeLeadChannel(SELF)
+                .withMailbox(PEER, LeadChannel.MailboxState.exists(PEER, 0, 0));
+
+        McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", null, null);
+
+        assertFalse(res.isError(), "a zero-consumer mailbox is still a successful, durably-queued publish");
+        String out = textOf(res);
+        assertTrue(out.contains("durably confirmed"), out);
+        assertTrue(out.toLowerCase().contains("no consumers"), () -> "must warn nobody is reading it: " + out);
+    }
+
+    @Test
+    void staysQuietAboutConsumersWhenThePeerMailboxHasOne() {
+        var channel = new FakeLeadChannel(SELF)
+                .withMailbox(PEER, LeadChannel.MailboxState.exists(PEER, 0, 1));
+
+        McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", null, null);
+
+        assertFalse(res.isError());
+        String out = textOf(res);
+        assertTrue(out.contains("durably confirmed"), out);
+        assertFalse(out.toLowerCase().contains("no consumers"), () -> "a consumer IS attached: " + out);
+    }
+
+    /**
+     * fleetd #361 review finding 1: an unmeasured fact must never render as a definite one. When
+     * the post-publish probe could not determine the mailbox's consumer count at all (broker slow,
+     * unreachable, or the probe timed out — {@link LeadChannel.MailboxState#unknown}), the result
+     * must stay just as quiet as the has-a-consumer case — never assert "no consumers" for a mailbox
+     * this call never actually measured.
+     */
+    @Test
+    void staysQuietAboutConsumersWhenThePeerMailboxStateIsUnknown() {
+        var channel = new FakeLeadChannel(SELF)
+                .withMailbox(PEER, LeadChannel.MailboxState.unknown(PEER));
+
+        McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", null, null);
+
+        assertFalse(res.isError(), "an unresolved post-publish probe must never turn a durably-confirmed publish into an error");
+        String out = textOf(res);
+        assertTrue(out.contains("durably confirmed"), out);
+        assertFalse(out.toLowerCase().contains("no consumers"),
+                () -> "an unmeasured fact must never be reported as a definite zero-consumer mailbox: " + out);
+    }
 }
@@ -10,6 +10,9 @@ import dev.ltms.fleet.herdr.FakeHerdr;
 import dev.ltms.fleet.herdr.PaneLocator;
 import dev.ltms.fleet.herdr.WorkspaceControl;
 import dev.ltms.fleet.inject.Injector;
+import dev.ltms.fleet.msg.FakeLeadChannel;
+import dev.ltms.fleet.msg.LeadChannel;
+import dev.ltms.fleet.msg.LeadMessage;
 import dev.ltms.fleet.msg.MessageService;
 import dev.ltms.fleet.msg.Rendezvous;
 import dev.ltms.fleet.session.FakeWorktrees;
@@ -34,7 +37,9 @@ import java.util.Map;
 import java.util.EnumSet;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;

 import static org.junit.jupiter.api.Assertions.*;
@@ -576,17 +581,48 @@ class FleetMcpTest {
    void listReportsThisDaemonsOwnCoordIdWhenLeadCoordinationIsOn() {
        FakeHerdr h = new FakeHerdr();
        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
+                .withMailbox("mac-opus", LeadChannel.MailboxState.exists("mac-opus", 0, 1));

        McpSchema.CallToolResult res = FleetMcp.listFleet(
                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
-                FleetMcp.QuarantineSource.none(), Map.of(), "", "mac-opus");
+                FleetMcp.QuarantineSource.none(), Map.of(), "",
+                new FleetMcp.CoordinationSource(channel, List.of()));

        String out = textOf(res);
-        // There is no peer-discovery surface yet, so this row answers the one question an operator
-        // cannot answer any other way: which coord-id a peer must use to reach ME.
+        // fleetd #361: reports both which coord-id a peer must use to reach ME, and this daemon's
+        // own mailbox state (a self-diagnosis: is my own consumer actually attached?).
        assertTrue(out.contains("\"coordinator\""), out);
        assertTrue(out.contains("\"selfId\":\"mac-opus\""), out);
+        assertTrue(out.contains("\"mailbox\":{\"status\":\"exists\",\"pending\":0,\"consumers\":1}"), out);
+        assertTrue(out.contains("\"held\":[]"), out);
+        assertTrue(out.contains("\"peers\":[]"), out);
+    }
+
+    /**
+     * fleetd #361 review finding 1: a self-probe that could not complete (broker unreachable, timed
+     * out) must never render the same as a measured "0 pending, 0 consumers" — that was exactly the
+     * bug: a reader could not tell "my mailbox is empty and idle" from "I could not check", and the
+     * second one is the far more alarming state.
+     */
+    @Test
+    void listReportsAnUnresolvedSelfProbeAsUnknownNeverAsAMeasuredZero() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
+                .withMailbox("mac-opus", LeadChannel.MailboxState.unknown("mac-opus"));
+
+        McpSchema.CallToolResult res = FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), Map.of(), "",
+                new FleetMcp.CoordinationSource(channel, List.of()));
+
+        String out = textOf(res);
+        assertTrue(out.contains("\"mailbox\":{\"status\":\"unknown\"}"), out);
+        assertFalse(out.contains("\"pending\""), "an unresolved probe must never carry a pending count at all: " + out);
+        assertFalse(out.contains("\"consumers\""), "an unresolved probe must never carry a consumers count at all: " + out);
    }

    @Test
@@ -601,6 +637,110 @@ class FleetMcpTest {
                "an ordinary fleet's output must be unchanged by this feature");
    }

+    @Test
+    void listReportsHeldMessagesWithATruncatedPreviewNeverTheFullBody() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        String longContent = "x".repeat(200);
+        FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
+                .hold(new LeadMessage("m1", "fleet01-lead", "mac-opus", longContent));
+
+        McpSchema.CallToolResult res = FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), Map.of(), "",
+                new FleetMcp.CoordinationSource(channel, List.of()));
+
+        String out = textOf(res);
+        assertTrue(out.contains("\"msgId\":\"m1\""), out);
+        assertTrue(out.contains("\"from\":\"fleet01-lead\""), out);
+        assertFalse(out.contains(longContent), "fleet_list must never dump a held message's full body: " + out);
+        assertTrue(out.contains("x".repeat(80) + "…"), "expected an 80-char preview with an ellipsis: " + out);
+    }
+
+    @Test
+    void listReportsEachDeclaredPeersLiveReachability() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
+                .withMailbox("fleet01-lead", LeadChannel.MailboxState.exists("fleet01-lead", 2, 1))
+                .withMailbox("fleet03-lead", LeadChannel.MailboxState.unknown("fleet03-lead"));
+        // "fleet02-lead" is declared as a peer but never configured on the fake — inspect() falls
+        // back to MailboxState.absent, exactly as a real down (never-run) peer would report.
+        // "fleet03-lead" IS configured, as unknown — a broker that could not be reached in time,
+        // which review finding 1 says must render distinctly from "fleet02-lead"'s confirmed absence.
+
+        McpSchema.CallToolResult res = FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), Map.of(), "",
+                new FleetMcp.CoordinationSource(channel, List.of("fleet01-lead", "fleet02-lead", "fleet03-lead")));
+
+        String out = textOf(res);
+        assertTrue(out.contains("\"coordId\":\"fleet01-lead\",\"status\":\"exists\",\"pending\":2,\"consumers\":1"), out);
+        assertTrue(out.contains("\"coordId\":\"fleet02-lead\",\"status\":\"absent\""), out);
+        assertTrue(out.contains("\"coordId\":\"fleet03-lead\",\"status\":\"unknown\""), out);
+        assertFalse(out.contains("\"coordId\":\"fleet02-lead\",\"status\":\"absent\",\"pending\""),
+                "pending/consumers must be omitted, not faked as zero, for a confirmed-absent peer: " + out);
+        assertFalse(out.contains("\"coordId\":\"fleet03-lead\",\"status\":\"unknown\",\"pending\""),
+                "pending/consumers must be omitted, not faked as zero, for an unresolved peer probe: " + out);
+    }
+
+    /**
+     * fleetd #361 review finding 2: {@code get(timeout)} alone times out the CALLER but leaves the
+     * submitted {@link LeadChannel#inspect} task running forever on its own virtual thread — against
+     * a hung (not down) broker every probe would orphan one more thread holding an AMQP channel
+     * until the connection's channel-max is exhausted, which would break {@code publish} too. This
+     * proves {@link FleetMcp#probe(LeadChannel, String, long)} does not merely give up on a slow
+     * task: it interrupts it, so the task does not go on running unbounded after the caller has
+     * already moved on. No hung broker needed — a {@link LeadChannel} fake that blocks until
+     * interrupted is enough to observe the same mechanism.
+     */
+    @Test
+    void aTimedOutProbeInterruptsTheOrphanedTaskRatherThanAbandoningIt() throws Exception {
+        CountDownLatch started = new CountDownLatch(1);
+        AtomicBoolean wasInterrupted = new AtomicBoolean(false);
+        LeadChannel hangs = new LeadChannel() {
+            @Override
+            public void publish(String toCoordId, LeadMessage m) { }
+
+            @Override
+            public List<LeadMessage> peek() { return List.of(); }
+
+            @Override
+            public void ack(String msgId) { }
+
+            @Override
+            public String selfCoordId() { return "mac-opus"; }
+
+            @Override
+            public MailboxState inspect(String coordId) {
+                started.countDown();
+                try {
+                    Thread.sleep(60_000);
+                } catch (InterruptedException e) {
+                    wasInterrupted.set(true);
+                    Thread.currentThread().interrupt();
+                }
+                return MailboxState.unknown(coordId);
+            }
+        };
+
+        LeadChannel.MailboxState result = FleetMcp.probe(hangs, "fleet01-lead", 100L);
+
+        assertFalse(result.exists(), "a timed-out probe must never claim the mailbox exists");
+        assertFalse(result.known(), "a timed-out probe proves nothing either way — it must report unknown");
+        assertTrue(started.await(2, TimeUnit.SECONDS), "the probe task must actually have started");
+        // The interrupt is delivered asynchronously to the orphaned task's own thread — poll briefly
+        // rather than assume it has already landed the instant probe() returns.
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+        while (!wasInterrupted.get() && System.nanoTime() < deadline) {
+            Thread.sleep(20);
+        }
+        assertTrue(wasInterrupted.get(),
+                "probe() must cancel the orphaned task (interrupt it) instead of leaving it to run forever");
+    }
+
    @Test
    void capacityUsesThePlacementLiveCount() {
        FakeHerdr h = new FakeHerdr();
@@ -911,7 +1051,8 @@ class FleetMcpTest {
        String out = textOf(FleetMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                sessions, null, new FleetMcp.CapacitySource(profile -> 2, profile -> 3,
                () -> Set.of("sonnet"), () -> 0), new FleetMcp.HealthCoverageSource(() -> "off"),
-                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), leadSeats, Map.of(), "", null));
+                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), leadSeats, Map.of(), "",
+                FleetMcp.CoordinationSource.none()));

        assertTrue(out.contains("\"maxLoad\":3"), "maxLoad itself must be left untouched: " + out);
        assertTrue(out.contains("\"live\":2"), out);
@@ -934,7 +1075,8 @@ class FleetMcpTest {
        String out = textOf(FleetMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                sessions, null, new FleetMcp.CapacitySource(profile -> 0, profile -> 3,
                () -> Set.of("sonnet"), () -> 0), new FleetMcp.HealthCoverageSource(() -> "off"),
-                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), leadSeats, Map.of(), "", null));
+                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), leadSeats, Map.of(), "",
+                FleetMcp.CoordinationSource.none()));

        assertTrue(out.contains("\"live\":0"), out);
        assertTrue(out.contains("\"free\":3"), "the real gate never subtracts the lead's seat: " + out);
@@ -988,7 +1130,7 @@ class FleetMcpTest {
        String out = textOf(FleetMcp.listFleet(composite, sm, null,
                new FleetMcp.CapacitySource(liveCount, p -> profiles.get(p).maxLoad(), profiles::keySet, () -> 0),
                new FleetMcp.HealthCoverageSource(() -> "off"), FleetMcp.QuarantineSource.none(),
-                FleetMcp.OutageSource.none(), leadSeats, Map.of(), "", null));
+                FleetMcp.OutageSource.none(), leadSeats, Map.of(), "", FleetMcp.CoordinationSource.none()));
        int reportedFree = extractInt(out, "free");

        for (int i = 0; i < reportedFree; i++) {
@@ -1017,7 +1159,7 @@ class FleetMcpTest {
                sessions, null, new FleetMcp.CapacitySource(profile -> 0, profile -> 2,
                () -> Set.of("terra"), () -> 0), new FleetMcp.HealthCoverageSource(() -> "off"),
                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", null));
+                Map.of(), "", FleetMcp.CoordinationSource.none()));

        assertTrue(out.contains("\"free\":2"), out);
        assertFalse(out.contains("leadSeats"), "no lead shares this profile's credential: " + out);
@@ -146,7 +146,7 @@ class MemberEnvAllowListTest {
        return new FleetConfig(null, null, null, Map.of(), null, null, null, null, null,
                new FleetConfig.Broker(null, brokerUriEnv, null), null, null, null, null, null,
                null, null, null, null,
-                new FleetConfig.Coordinator(null, coordinatorUriEnv, null, null)).withDefaults();
+                new FleetConfig.Coordinator(null, coordinatorUriEnv, null, null, null)).withDefaults();
    }

    /** {@code LC_*} categories are infrastructure by prefix; everything else needs an exact match. */
@@ -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);
+    }
+}
@@ -3,6 +3,8 @@ package dev.ltms.fleet.msg;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;

 /**
 * Hermetic stand-in for {@link LeadChannel}: an in-memory mailbox that records what was published
@@ -24,11 +26,19 @@ public final class FakeLeadChannel implements LeadChannel {
    private final List<String> acked = Collections.synchronizedList(new ArrayList<>());
    /** When set, every {@link #publish} throws it — the unroutable/nacked/timed-out peer. */
    private volatile IllegalStateException publishFailure;
+    /** Canned {@link #inspect} results by coord-id — absent for any coord-id not configured here. */
+    private final Map<String, MailboxState> mailboxes = new ConcurrentHashMap<>();

    public FakeLeadChannel(String selfCoordId) {
        this.selfCoordId = selfCoordId;
    }

+    /** Make {@link #inspect(String)} return {@code state} for {@code coordId} instead of "absent". */
+    public FakeLeadChannel withMailbox(String coordId, MailboxState state) {
+        mailboxes.put(coordId, state);
+        return this;
+    }
+
    /** Make every publish fail as an unreachable peer would. */
    public FakeLeadChannel failPublishWith(String message) {
        this.publishFailure = new IllegalStateException(message);
@@ -65,6 +75,11 @@ public final class FakeLeadChannel implements LeadChannel {
        return selfCoordId;
    }

+    @Override
+    public MailboxState inspect(String coordId) {
+        return mailboxes.getOrDefault(coordId, MailboxState.absent(coordId));
+    }
+
    public List<LeadMessage> published() {
        return List.copyOf(published);
    }
@@ -0,0 +1,77 @@
+package dev.ltms.fleet.msg;
+
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQImpl;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #361 review round 2: a mutation that made {@link LeadMailbox#isMissingQueue} return
+ * {@code true} unconditionally still left {@code mvn clean install} green — 1389 tests, 0
+ * failures — because nothing exercised its false branch. That branch is the whole discriminator
+ * between {@link LeadChannel.MailboxState#absent} and {@link LeadChannel.MailboxState#unknown};
+ * without a test pinning it, a future refactor that widens it back to "always true" (restoring the
+ * exact overstatement fleetd #361 exists to fix) would pass this suite.
+ *
+ * <p>Hermetic — no broker needed, per the review's own suggestion. {@code isMissingQueue} takes a
+ * plain {@link IOException}, so every input here is constructed directly rather than provoked from
+ * a live connection. The real 404 shape itself is still pinned against a real broker, in
+ * {@code LeadMailboxTest.passiveDeclareOfAMissingQueueThrowsAnIOExceptionWrappingA404ShutdownSignal}
+ * — this class covers the three false shapes {@link LeadMailbox#isMissingQueue}'s own javadoc
+ * lists, so both directions of the discriminator are proven somewhere.
+ */
+class LeadMailboxIsMissingQueueTest {
+
+    @Test
+    void aConfirmedMissingQueueIsRecognized() {
+        ShutdownSignalException sse = new ShutdownSignalException(true, false,
+                new AMQImpl.Channel.Close(404, "NOT_FOUND - no queue 'lead.x.inbox' in vhost '/'", 50, 10), null);
+        IOException e = new IOException("channel error", sse);
+
+        assertTrue(LeadMailbox.isMissingQueue(e), "a genuine 404 Channel.Close must be recognized as a missing queue");
+    }
+
+    @Test
+    void aDifferentReplyCodeIsNotAMissingQueue() {
+        // E.g. 403 ACCESS_REFUSED — the queue may well exist; this call was simply refused.
+        ShutdownSignalException sse = new ShutdownSignalException(true, false,
+                new AMQImpl.Channel.Close(403, "ACCESS_REFUSED", 50, 10), null);
+        IOException e = new IOException("channel error", sse);
+
+        assertFalse(LeadMailbox.isMissingQueue(e),
+                "a non-404 reply code must never be read as a confirmed absence — the mailbox's real state is unknown");
+    }
+
+    @Test
+    void aShutdownSignalWhoseReasonIsNotAChannelCloseIsNotAMissingQueue() {
+        // A Connection.Close (a whole different broker-level shutdown) is still a ShutdownSignalException,
+        // but its reason is not a Channel.Close at all — must not be misread as "no such queue".
+        ShutdownSignalException sse = new ShutdownSignalException(true, false,
+                new AMQImpl.Connection.Close(404, "coincidentally 404, but this is a CONNECTION close", 10, 50), null);
+        IOException e = new IOException("connection error", sse);
+
+        assertFalse(LeadMailbox.isMissingQueue(e),
+                "a ShutdownSignalException whose reason is not a Channel.Close must never be read as a missing queue,"
+                        + " even if its reply code happens to be 404");
+    }
+
+    @Test
+    void anIOExceptionWithNoCauseAtAllIsNotAMissingQueue() {
+        IOException e = new IOException("some other declare failure, no cause attached");
+
+        assertFalse(LeadMailbox.isMissingQueue(e),
+                "an IOException with no ShutdownSignalException cause must never be read as a confirmed absence");
+    }
+
+    @Test
+    void anIOExceptionWithAnUnrelatedCauseIsNotAMissingQueue() {
+        IOException e = new IOException("wrapped something else entirely", new RuntimeException("boom"));
+
+        assertFalse(LeadMailbox.isMissingQueue(e),
+                "a cause that isn't even a ShutdownSignalException must never be read as a confirmed absence");
+    }
+}
@@ -1,5 +1,9 @@
 package dev.ltms.fleet.msg;

+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ShutdownSignalException;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
@@ -7,11 +11,14 @@ import org.testcontainers.containers.RabbitMQContainer;
 import org.testcontainers.junit.jupiter.Testcontainers;
 import org.testcontainers.utility.DockerImageName;

+import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;

 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.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;

@@ -159,6 +166,145 @@ class LeadMailboxTest {
        }
    }

+    @Test
+    void inspectReportsAnOwnedMailboxAsExistingWithItsOwnConsumer() throws Exception {
+        // A LeadMailbox declares AND consumes its own queue the moment open() returns (see own()),
+        // so inspecting a coord-id this same process owns must always find exactly one consumer.
+        String self = coordId("lead-inspect-self");
+        try (LeadMailbox mailbox = LeadMailbox.open(uri(), self)) {
+            LeadChannel.MailboxState state = mailbox.inspect(self);
+            assertEquals(self, state.coordId());
+            assertTrue(state.exists(), "this daemon owns and has declared this exact queue");
+            assertEquals(0, state.pending(), "nothing has been published to it yet");
+            assertEquals(1, state.consumers(), "the mailbox's own constructor already attached a consumer");
+        }
+    }
+
+    @Test
+    void inspectReportsAMissingMailboxAsAbsentRatherThanThrowing() throws Exception {
+        String nobody = coordId("lead-inspect-nobody");
+        try (LeadMailbox mailbox = LeadMailbox.open(uri(), coordId("lead-inspect-caller"))) {
+            LeadChannel.MailboxState state = mailbox.inspect(nobody);
+            assertEquals(LeadChannel.MailboxState.absent(nobody), state,
+                    "a queue nobody has ever declared must report absent, never throw");
+            assertTrue(state.known(), "a confirmed 404 IS a definite answer — this is not the unknown case");
+        }
+    }
+
+    /**
+     * fleetd #361 review finding 3: {@code inspect} is specified to never throw, but the original
+     * implementation caught only {@link IOException} — and {@link Connection#createChannel()} on an
+     * already-closed connection throws {@link com.rabbitmq.client.AlreadyClosedException}, an
+     * unchecked {@link RuntimeException} (pinned by {@code
+     * createChannelOnAnAlreadyClosedConnectionThrowsAnUncheckedException} above). This drives that
+     * exact scenario through the real {@link LeadMailbox#inspect} — not the raw client call — and
+     * checks both halves of finding 1 and finding 3 at once: no exception escapes, and the result is
+     * {@code UNKNOWN} rather than the wrong-but-plausible-looking {@code ABSENT}.
+     */
+    @Test
+    void inspectReportsUnknownRatherThanThrowingWhenTheConnectionIsAlreadyClosed() throws Exception {
+        LeadMailbox mailbox = LeadMailbox.open(uri(), coordId("lead-inspect-dead-connection"));
+        mailbox.close(); // tears down the connection `inspect` will try to open a probe channel on
+
+        LeadChannel.MailboxState state = mailbox.inspect(coordId("lead-inspect-irrelevant-target"));
+
+        assertFalse(state.exists());
+        assertFalse(state.known(), "a dead connection proves nothing about the target mailbox — it must be unknown, not absent");
+        assertEquals(LeadChannel.MailboxState.Presence.UNKNOWN, state.presence());
+    }
+
+    @Test
+    void inspectReportsPendingMessagesAndZeroConsumersWhenNobodyIsReadingAnymore() throws Exception {
+        // Publish into a mailbox this test owns, then never consume from it, to prove `pending` and
+        // `consumers` really come off the broker rather than off this process's own in-memory state.
+        String to = coordId("lead-inspect-pending");
+        String observerId = coordId("lead-inspect-observer");
+        try (LeadMailbox owner = LeadMailbox.open(uri(), to);
+             LeadMailbox observer = LeadMailbox.open(uri(), observerId)) {
+            owner.publish(to, new LeadMessage("m1", "lead-from", to, "sitting in the queue"));
+            awaitPeek(owner); // make sure the broker has actually enqueued it before inspecting
+        } // `owner` closes here: its consumer disconnects, but the durable, unacked message stays queued.
+
+        try (LeadMailbox observer = LeadMailbox.open(uri(), coordId("lead-inspect-observer-2"))) {
+            // The broker requeues `owner`'s unacked delivery asynchronously once its connection drops,
+            // so poll rather than assume the very first passive declare already sees the settled state.
+            LeadChannel.MailboxState state = awaitInspect(observer, to, s -> s.consumers() == 0);
+            assertTrue(state.exists());
+            assertEquals(0, state.consumers(), "the only owner just closed — nobody is reading this anymore");
+            assertEquals(1, state.pending(), "the unacked message must be requeued, never dropped");
+        }
+    }
+
+    /**
+     * fleetd #361's central invariant, proved rather than assumed: a passive queue declare of a
+     * missing queue closes ITS channel with a 404 in AMQP 0-9-1. {@link LeadMailbox#inspect} is
+     * specified to run on its own disposable channel for exactly this reason — this test is the one
+     * that actually exercises the failure mode and shows {@link LeadMailbox#publish} on the SAME
+     * instance is unaffected by it.
+     */
+    @Test
+    void inspectingAMissingMailboxNeverBreaksPublishOnTheSameInstance() throws Exception {
+        String self = coordId("lead-invariant-self");
+        try (LeadMailbox mailbox = LeadMailbox.open(uri(), self)) {
+            // Miss on a queue that has never existed — this is exactly the 404-closes-the-channel case.
+            LeadChannel.MailboxState missed = mailbox.inspect(coordId("lead-invariant-nobody-home"));
+            assertFalse(missed.exists());
+            assertTrue(missed.known(), "a genuine 404 on a queue that never existed is a confirmed fact, not an unknown");
+
+            // publish() must still work on THIS SAME instance: if inspect() had reused `publishChannel`
+            // (or `channel`), the broker's 404 would have closed it out from underneath publish().
+            LeadMessage sent = new LeadMessage("after-miss", "lead-from", self, "still alive");
+            mailbox.publish(self, sent);
+            List<LeadMessage> got = awaitPeek(mailbox);
+            assertEquals(1, got.size(), "publish must still reach this mailbox's own queue after a missed inspect");
+            assertEquals("after-miss", got.getFirst().msgId());
+
+            // And a second inspect() — of a mailbox that DOES exist this time — must also still work,
+            // proving the miss did not wedge inspect() itself either.
+            LeadChannel.MailboxState self2 = mailbox.inspect(self);
+            assertTrue(self2.exists());
+        }
+    }
+
+    /**
+     * Pins the exact exception shape {@link LeadMailbox#inspect} relies on to tell a genuine 404
+     * (mailbox confirmed absent) apart from everything else (mailbox state unknown) — measured
+     * against a real broker rather than assumed from the AMQP 0-9-1 spec text. If this ever fails,
+     * the classification in {@code inspect} is reading the wrong shape and must be revisited.
+     */
+    @Test
+    void passiveDeclareOfAMissingQueueThrowsAnIOExceptionWrappingA404ShutdownSignal() throws Exception {
+        try (Connection conn = LeadMailbox.connectionFactory(uri()).newConnection()) {
+            Channel probe = conn.createChannel();
+            String missing = LeadMailbox.queueName(coordId("lead-404-shape"));
+            IOException thrown = assertThrows(IOException.class, () -> probe.queueDeclarePassive(missing));
+            assertInstanceOf(ShutdownSignalException.class, thrown.getCause(),
+                    () -> "expected the IOException to wrap a ShutdownSignalException, got: " + thrown);
+            ShutdownSignalException sse = (ShutdownSignalException) thrown.getCause();
+            assertInstanceOf(AMQP.Channel.Close.class, sse.getReason(),
+                    () -> "expected a Channel.Close reason: " + sse);
+            AMQP.Channel.Close close = (AMQP.Channel.Close) sse.getReason();
+            assertEquals(404, close.getReplyCode(), () -> "expected AMQP NOT_FOUND (404): " + close);
+            assertFalse(probe.isOpen(), "the 404 must have closed the channel the declare ran on");
+        }
+    }
+
+    /**
+     * The other half of the same measurement: calling {@code createChannel()} on an
+     * already-closed connection — the shape {@link LeadMailbox#inspect} hits when the broker
+     * connection itself is gone — throws {@link com.rabbitmq.client.AlreadyClosedException}, an
+     * unchecked {@link RuntimeException}, not an {@link IOException}. An {@code inspect} that only
+     * caught {@code IOException} here would let this escape instead of reporting "unknown".
+     */
+    @Test
+    void createChannelOnAnAlreadyClosedConnectionThrowsAnUncheckedException() throws Exception {
+        Connection conn = LeadMailbox.connectionFactory(uri()).newConnection();
+        conn.close();
+        RuntimeException thrown = assertThrows(RuntimeException.class, conn::createChannel);
+        assertInstanceOf(com.rabbitmq.client.AlreadyClosedException.class, thrown,
+                () -> "expected AlreadyClosedException, got: " + thrown);
+    }
+
    /** Poll peek until at least one message is held, or ~10s elapse (broker delivery is async). */
    @SuppressWarnings("BusyWait")
    private static List<LeadMessage> awaitPeek(LeadMailbox inbox) throws InterruptedException {
@@ -170,4 +316,18 @@ class LeadMailboxTest {
        }
        return msgs;
    }
+
+    /** Poll inspect(coordId) until it satisfies {@code done}, or ~10s elapse (broker state settles async). */
+    @SuppressWarnings("BusyWait")
+    private static LeadChannel.MailboxState awaitInspect(
+            LeadMailbox observer, String coordId, java.util.function.Predicate<LeadChannel.MailboxState> done)
+            throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        LeadChannel.MailboxState state = observer.inspect(coordId);
+        while (!done.test(state) && System.nanoTime() < deadline) {
+            Thread.sleep(50);
+            state = observer.inspect(coordId);
+        }
+        return state;
+    }
 }
@@ -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"
@@ -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'
Author	SHA1	Message	Date
Dai Ha	01492059d4	fleetd #361 review round 2: pin isMissingQueue's false branch CI / contract (pull_request) Successful in 1m5s Details CI / build (pull_request) Successful in 1m34s Details The reviewer's mutation (isMissingQueue always returns true) restored the exact overstatement fleetd #361 exists to fix -- every declare failure reading as a confirmed absence -- and still left mvn clean install green (1389/1389), because no test drove a non-404 shape through inspect(). The false branch was the whole discriminator between MailboxState.absent() and MailboxState.unknown(), unpinned. Widened LeadMailbox.isMissingQueue from private to package-private and added LeadMailboxIsMissingQueueTest: five hermetic tests (no broker) covering the true case and all three false shapes isMissingQueue's own javadoc lists -- a different reply code, a ShutdownSignalException whose reason isn't a Channel.Close, and an IOException with no such cause at all (plus an IOException wrapping an unrelated exception type). Re-ran the reviewer's exact mutation locally: 4 of 5 new tests went red with the expected assertion messages; reverted, and mvn clean install is green again at 1394/1394 (1389 + 5 new). Also added a one-line javadoc note on LeadMailbox.inspect being honest about which of its two RuntimeException catches is proven by a test (the createChannel() one, end-to-end against a real broker) and which stays purely defensive (the declare-site one, for a connection-drops- mid-call race no test drives on purpose).	2026-09-05 13:12:24 +07:00
Dai Ha	c4d40fbc2b	fleetd #361 review: fix false-negative absent, uncancelled probes, and a throw contract gap CI / contract (pull_request) Successful in 47s Details CI / build (pull_request) Successful in 1m45s Details Three findings from review of #364, fixed on the same branch: 1. LeadChannel.MailboxState.absent() was returned both for a genuinely absent mailbox AND for "the probe could not determine anything" (timeout, unreachable broker, other declare failure) -- exactly the overstatement #361 exists to fix, one level down. MailboxState now carries a Presence enum (EXISTS/ABSENT/UNKNOWN) with exists()/known() accessors; LeadMailbox.inspect classifies a real AMQP 404 (measured against a live broker, not assumed: an IOException wrapping a ShutdownSignalException whose Channel.Close reply code is 404) as ABSENT and everything else as UNKNOWN. fleet_list's mailbox/peer rows now render a "status" of exists/absent/unknown and only include pending/consumers when status is "exists", so an unresolved self- or peer-probe can never render as a measured zero. 2. FleetMcp.probe's get(timeoutMs) left a timed-out inspect() task running forever on its own virtual thread, holding the AMQP channel it had already opened -- against a hung (not down) broker this would orphan one channel per fleet_list call until the connection's channel-max was exhausted, breaking publish() too. probe() now holds the Future and calls cancel(true) on timeout/failure so the orphaned task is interrupted instead of abandoned, and now returns MailboxState.unknown() (never absent()) on timeout/exception. 3. LeadMailbox.inspect only caught IOException, but createChannel() on an already-closed connection throws AlreadyClosedException, an unchecked RuntimeException (measured against a live broker) -- so it could escape the "never throws" contract. Both places in inspect now also catch RuntimeException and report unknown(). Tests: MailboxState.exists()/absent()/unknown() call sites updated across FleetMcpTest/FleetMcpLeadCoordTest; new hermetic tests cover the tri-state fleet_list rendering (self-probe unknown, a peer that's absent vs. one that's unknown) and probe cancellation (a LeadChannel fake that blocks until interrupted, proving probe() doesn't just give up on it); new @Tag("contract") LeadMailboxTest cases pin the real exception shapes for both the 404 and the already-closed-connection paths and prove inspect() reports unknown (never throws) when the connection is already closed.	2026-09-05 13:01:48 +07:00
Dai Ha	0df34f3220	fleetd #361 : close the lead-coordination visibility gap CI / contract (pull_request) Successful in 1m10s Details CI / build (pull_request) Successful in 1m34s Details Lead-to-lead AMQP coordination had a send half with tools and a receive half without. This closes three blind spots: - LeadChannel gains inspect(coordId) -> MailboxState(exists, pending, consumers), implemented in LeadMailbox with a throwaway probe channel (never the long-lived publish/consume channels) so a passive-declare 404 on a missing queue can never take down publish() on the same instance. - FleetConfig.Coordinator gains peers: List<String> (defaults to empty, blank entries dropped) so a daemon can declare which peer coord-ids it expects to reach. - fleet_list reports coordination state via a new CoordinationSource (own coord-id, own mailbox state, held messages as msgId/from/preview only, and one row per configured peer with reachability/pending/ consumers), following the existing OutageSource/QuarantineSource "Source record with none()" idiom instead of growing listFleet's overload chain by another positional parameter. Every peer probe is bounded by a 1.5s timeout on a virtual-thread pool and degrades to absent rather than ever slowing or failing fleet_list. - fleet_send{coordId}'s success text now says "durably confirmed by the broker" instead of "delivered", and warns (while still reporting success) when the target mailbox has zero consumers attached. Tests: hermetic unit tests for exists/absent/zero-consumer/old-config- no-peers-key/new fleet_list shape using FakeLeadChannel, plus a @Tag("contract") LeadMailboxTest.inspectingAMissingMailboxNeverBreaks PublishOnTheSameInstance proving the invariant against a real broker.	2026-09-05 12:44:29 +07:00
Dai Ha	3759c41f99	Merge #354 : the redeploy health gate classifies AMQP errors instead of counting them CI / contract (push) Successful in 1m14s Details CI / build (push) Successful in 1m36s Details The gate counted ERROR lines since RESTART_MARK. On a laptop that idle-sleeps after one minute on battery that meant 6 ERROR lines for an AMQP link that recovered every time, and a gate that cries wolf is a gate nobody reads. It now reports three states: no errors; only errors proven to have recovered (quiet, and the gate passes); anything else (the old warning, unchanged). Attribution is per connection, using the names #356 put into the log -- a lead-mailbox recovery can no longer clear an unrecovered reply-inbox reset. A candidate carrying neither name is unattributable and stays LOUD. Two earlier rounds were rejected. Round 1 was inert: it matched nothing in the real log, because the layout abbreviates the logger and 'Connection reset' sits in the stack trace, not on the ERROR line -- my brief had pointed the worker at fleetd.out, which is untracked and so absent from its worktree. Round 2 was correct and honest but could not attribute anything, which is what motivated #356. Verified on merge beyond the worker's own mutations: - ran the classifier against the REAL log, which is still in the pre-#356 format: 6 total, 0 recovered, 6 unexplained. Old-format lines carry no connection name, so they stay loud -- the safe direction, on genuine data rather than a fixture. - adversarial fixture the worker did not write: a lead recovery BEFORE any failure banks no credit; 2 inbox resets with 1 recovery leaves 1 unexplained; a non-AMQP ERROR stays loud. total=3 recovered=1 unexplained=2, as intended. - RESTART_MARK still anchors the scanned region. Caveat carried from the PR: the patterns are source-derived. The daemon has not been redeployed, so they are not yet confirmed against a live log.	2026-09-05 06:09:29 +07:00
Dai Ha	09159f2857	Classify named AMQP recovery errors CI / contract (pull_request) Successful in 51s Details CI / build (pull_request) Successful in 2m16s Details	2026-09-05 06:05:52 +07:00
Dai Ha	29cd1194c2	Merge remote-tracking branch 'origin/main' into worker/errscan-bed2ca-2	2026-09-05 06:02:29 +07:00
Dai Ha	815e8f8b23	Merge #356 : name the AMQP connection in its own log lines CI / contract (push) Successful in 45s Details CI / build (push) Successful in 1m38s Details 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.	2026-09-05 06:01:30 +07:00
Dai Ha	1e60ac0745	merge main for verification	2026-09-05 05:59:22 +07:00
Dai Ha	650a4c146b	Merge #357 : a FleetConfig component dropped by withDefaults() now fails the build CI / contract (push) Successful in 53s Details CI / build (push) Successful in 1m44s Details Test-only. FleetConfig.java itself is unchanged. The hazard is the back-compat constructor ladder (21/20/18/17/16/15/14 alongside the 22-arg canonical). Add a component and leave withDefaults()'s call at the old arity and it binds to a back-compat constructor: it compiles, the suite passes, and the new key is silently defaulted away on every load(). Verified on merge with a mutation the worker did not run: I made withDefaults() issue a 21-arg call, reproducing the real binding rather than an explicit null. It compiled, and the guard failed by name -- 'memberLoginShell: ... a component silently dropped by withDefaults(), the shape of the defect this test exists to catch'. Exclusion list is empty and its size is pinned, so a future exemption must touch an assertion rather than grow quietly.	2026-09-05 05:55:22 +07:00
Dai Ha	23f299e105	Preserve strict AMQP exception handling CI / contract (pull_request) Successful in 1m4s Details CI / build (pull_request) Successful in 1m46s Details	2026-09-05 05:53:44 +07:00
Dai Ha	d292522d00	Name AMQP connection failure logs CI / contract (pull_request) Successful in 1m20s Details CI / build (pull_request) Successful in 1m29s Details	2026-09-05 05:45:37 +07:00
Dai Ha	0241e0d3a8	Keep unattributed AMQP errors loud CI / contract (pull_request) Successful in 52s Details CI / build (pull_request) Successful in 1m30s Details	2026-09-05 05:36:53 +07:00
Dai Ha	e4973eb8a4	Classify recovered AMQP redeploy errors CI / contract (pull_request) Successful in 45s Details CI / build (pull_request) Successful in 1m51s Details	2026-09-05 05:29:53 +07:00