fleetd #512 (part 1): log a positive completion line when drainAll finishes

drainAll used to log nothing on a clean drain — both existing log calls
(drainSnapshot's per-session failure, drainAll's straggler-sweep warning)
sit on abnormal paths, so "drained fine" and "died on the first session"
looked identical: no log line either way.

Add one log.info at the end of drainAll: "drain complete: released=N
abandoned=M (still BUSY at the shutdown deadline)". It fires on the
normal path, including the all-zero case, and folds both drainSnapshot
passes (main snapshot + straggler sweep) into one line.

drainSnapshot now returns a private DrainTally(released, abandoned)
record instead of void, and the private release(paneId, cause) overload
now returns the removed MemberSession (previously void) so drainSnapshot
can read its state at the moment of removal — the same check
logPreservedForShutdown already makes. Both signature changes are
private with a single call site, so the blast radius stays small.

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

@@ -244,7 +244,15 @@ public final class CallerResolver {
         // already names what happens if that case is handed the primary role: a worker→primary
         // escalation. So an unresolved caller is refused (ANONYMOUS — the same clean, already-tested
         // "authenticated as nothing" outcome used everywhere else in this method), never promoted.
-        return isLoopback(remoteAddr) && c.resolved() ? Principal.primary(c.pid()) : Principal.anonymous();
+        //
+        // fleetd #505: the OTHER way a real pid can wrongly reach here with a null terminal — not a
+        // failed lsof lookup, but a herdr error partway through PaneLocator's pane scan. c.resolved()
+        // says nothing about that; it only tests the lsof sentinel (by design — see
+        // ConnectionIdentity.Caller#resolved). c.scanComplete() is the separate signal: a scan that
+        // could not check every pane must not be read as "checked everywhere, no match" — the pane it
+        // could not check might have been the caller's own. So both must hold before this promotes.
+        return isLoopback(remoteAddr) && c.resolved() && c.scanComplete()
+                ? Principal.primary(c.pid()) : Principal.anonymous();
     }
 
     private boolean presentedTokenMatches(String authorizationHeader) {

fleetd/src/main/java/dev/ltms/fleet/herdr/PaneLocator.java

@@ -1,6 +1,8 @@
 package dev.ltms.fleet.herdr;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -36,6 +38,8 @@ import java.util.Set;
  */
 public final class PaneLocator {
 
+    private static final Logger log = LoggerFactory.getLogger(PaneLocator.class);
+
     /**
      * Bound on how many ancestor generations {@link #ancestorsOf} walks. This runs on every MCP
      * call, so a cycle or a pathologically deep process tree must not hang identity resolution;
@@ -73,22 +77,46 @@ public final class PaneLocator {
     }
 
     /**
-     * The {@code terminal_id} of the agent pane whose process tree contains {@code pid}, or
-     * {@code null} if no agent pane on any searched daemon owns it (e.g. the caller is the
-     * primary, or off-host).
+     * The outcome of a {@link #terminalForPid} scan: the {@code terminal_id} of the agent pane
+     * whose process tree contains the pid ({@link #terminal} is {@code null} if none matched),
+     * and whether the scan that produced that answer ran to completion on every daemon searched.
+     *
+     * <p>{@link #complete} is {@code false} exactly when some {@code pane.process_info} call
+     * failed and, despite that, no pane was ever found to own the pid. In that case a {@code null}
+     * {@link #terminal} means "could not tell", not "definitely not a worker" — fleetd #505: a
+     * transient herdr error on the very pane that <em>does</em> own the caller's pid must not read
+     * as a clean negative and fall through to {@code Principal.primary}, the same way #317's
+     * {@code Caller.resolved()} already guards a failed lsof lookup. Callers ({@code
+     * ConnectionIdentity}, {@code CallerResolver}) must refuse rather than promote on an incomplete
+     * scan.
+     *
+     * <p>When a pane genuinely owns the pid, {@link #complete} is {@code true} regardless of
+     * whether some other, unrelated pane failed to answer earlier in the same scan — a positive
+     * match is definitive and does not need every pane to have been checked (a pane that "vanished
+     * mid-scan" but was never the match is still a clean, complete result).
      */
-    public String terminalForPid(long pid) {
+    public record Lookup(String terminal, boolean complete) {
+        private static final Lookup NOT_FOUND = new Lookup(null, true);
+    }
+
+    /**
+     * Resolve {@code pid} to the agent pane whose process tree contains it, across every searched
+     * herdr daemon. See {@link Lookup} for how to read a {@code null} terminal.
+     */
+    public Lookup terminalForPid(long pid) {
         if (pid <= 0) {
-            return null;
+            return Lookup.NOT_FOUND;
         }
         Set<Long> ancestry = ancestorsOf(pid);
-        for (HerdrClient herdr : herdrs) {
-            String terminal = terminalForPid(herdr, ancestry);
-            if (terminal != null) {
-                return terminal;
+        boolean complete = true;
+        for (int i = 0; i < herdrs.size(); i++) {
+            Lookup outcome = scan(herdrs.get(i), i, herdrs.size(), ancestry);
+            if (outcome.terminal() != null) {
+                return outcome; // a definite match — no need to finish checking other clients
             }
+            complete = complete && outcome.complete();
         }
-        return null;
+        return new Lookup(null, complete);
     }
 
     /**
@@ -117,31 +145,51 @@ public final class PaneLocator {
         return ancestry;
     }
 
-    private static String terminalForPid(HerdrClient herdr, Set<Long> ancestry) {
+    /** Whether a pane owns one of the scanned pid's ancestors, or the check of it failed outright. */
+    private enum Ownership { OWNS, DOES_NOT_OWN, UNKNOWN }
+
+    private static Lookup scan(HerdrClient herdr, int clientIndex, int clientCount, Set<Long> ancestry) {
+        boolean complete = true;
         for (JsonNode pane : herdr.call("pane.list", Map.of()).path("panes")) {
             String paneId = pane.path("pane_id").asText(null);
-            if (paneId != null && paneOwnsAnyOf(herdr, paneId, ancestry)) {
-                return pane.path("terminal_id").asText(null);
+            if (paneId == null) {
+                continue;
+            }
+            Ownership owns = paneOwnsAnyOf(herdr, clientIndex, clientCount, paneId, ancestry);
+            if (owns == Ownership.OWNS) {
+                return new Lookup(pane.path("terminal_id").asText(null), true);
+            }
+            if (owns == Ownership.UNKNOWN) {
+                complete = false;
             }
         }
-        return null;
+        return new Lookup(null, complete);
     }
 
-    private static boolean paneOwnsAnyOf(HerdrClient herdr, String paneId, Set<Long> ancestry) {
+    private static Ownership paneOwnsAnyOf(HerdrClient herdr, int clientIndex, int clientCount,
+                                            String paneId, Set<Long> ancestry) {
         JsonNode info;
         try {
             info = herdr.call("pane.process_info", Map.of("pane_id", paneId)).path("process_info");
         } catch (HerdrException e) {
-            return false; // pane vanished mid-scan — just skip it
+            // fleetd #505: this used to be read as a clean "does not own it" (the pane vanished
+            // mid-scan, just skip it) — one boolean carrying two different facts. It is UNKNOWN
+            // now: if THIS pane is the one that owns the pid, the caller must not be told "no pane
+            // owns it", because that reads as a real primary and is promoted under loopback-trust.
+            log.warn("pane.process_info failed for pane {} on herdr client {} of {} during a "
+                            + "pid-owner scan — treating it as \"could not tell\", not a clean "
+                            + "negative (fleetd #505): {}",
+                    paneId, clientIndex + 1, clientCount, e.getMessage());
+            return Ownership.UNKNOWN;
         }
         if (ancestry.contains(info.path("shell_pid").asLong(-1))) {
-            return true;
+            return Ownership.OWNS;
         }
         for (JsonNode p : info.path("foreground_processes")) {
             if (ancestry.contains(p.path("pid").asLong(-1))) {
-                return true;
+                return Ownership.OWNS;
             }
         }
-        return false;
+        return Ownership.DOES_NOT_OWN;
     }
 }

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

@@ -33,9 +33,10 @@ public final class ConnectionIdentity {
 
     /**
      * The caller resolved from the connection: its worker {@code terminal} (or {@code null} for the
-     * primary / an off-host client) and its {@code pid} (or {@code -1} if not resolvable).
+     * primary / an off-host client), its {@code pid} (or {@code -1} if not resolvable), and whether
+     * the pane scan behind {@code terminal} ran to completion ({@link #scanComplete}).
      */
-    public record Caller(String terminal, long pid) {
+    public record Caller(String terminal, long pid, boolean scanComplete) {
 
         /**
          * Whether the OS peer-PID lookup actually succeeded — {@code false} means {@code pid} is
@@ -51,6 +52,10 @@ public final class ConnectionIdentity {
          * {@link ConnectionIdentity#isLoopback} is centralised rather than left for each caller to
          * reimplement: a raw {@code pid > 0} check duplicated at every call site is precisely the
          * "one rule, two copies" shape that let #305 drift.
+         *
+         * <p>This method is deliberately NOT widened for fleetd #505's failure (a herdr error
+         * during the pane scan, not a failed lsof lookup) — it still tests only the sentinel it is
+         * named for. #505 is a different axis, carried separately in {@link #scanComplete}.
          */
         public boolean resolved() {
             return pid > 0;
@@ -60,10 +65,11 @@ public final class ConnectionIdentity {
     /** Resolve the caller's terminal and PID from one peer-PID lookup. */
     public Caller resolve(String remoteAddr, int remotePort) {
         if (!isLoopback(remoteAddr)) {
-            return new Caller(null, -1); // only same-host callers can be workers
+            return new Caller(null, -1, true); // only same-host callers can be workers
         }
         long pid = pids.pidForLocalPort(remotePort);
-        return new Caller(panes.terminalForPid(pid), pid);
+        PaneLocator.Lookup lookup = panes.terminalForPid(pid);
+        return new Caller(lookup.terminal(), pid, lookup.complete());
     }
 
     /**

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

@@ -527,14 +527,28 @@ public final class FleetMcp {
     }
 
     /**
-     * Pre-CB-501 identity: worker if the connection maps to a pane, otherwise the primary. Used
-     * only by the legacy constructor, where authorization is not enforced anyway.
+     * Pre-CB-501 identity: worker if the connection maps to a pane, otherwise anonymous. Used
+     * only by the legacy constructor ({@code callers == null}), where authorization is not
+     * enforced anyway — but the resolved {@link Principal} still reaches non-authz logic (e.g.
+     * {@code markSpawnedMemberPresent}, {@code recordPrimarySingleton}), so it must not be trusted
+     * with a role it did not earn.
+     *
+     * <p>fleetd #509: this used to fall back to {@link Principal#primary}, unconditionally, for
+     * every caller the connection did not resolve to a worker pane — with none of
+     * {@code CallerResolver.java:254}'s two guards ({@code isLoopback}, {@code scanComplete}).
+     * That is the exact shape #317 and #505 each closed on the enforced path; this branch was the
+     * same trap, left open on the legacy one. It now returns {@link Principal#anonymous} instead,
+     * so an unresolved legacy caller earns no authority rather than the primary's.
+     *
+     * <p>Package-private (was {@code private}) so this is unit-testable directly, the same reason
+     * {@link #denyFor} was split out — it runs inside a contextExtractor closure that only fires on
+     * a real MCP request, so nothing else could pin this behaviour.
      */
-    private static Principal legacyPrincipal(ConnectionIdentity identity, String addr, int port) {
+    static Principal legacyPrincipal(ConnectionIdentity identity, String addr, int port) {
         ConnectionIdentity.Caller c = identity.resolve(addr, port);
         return c.terminal() != null
                 ? Principal.worker(c.terminal(), c.pid())
-                : Principal.primary(c.pid());
+                : Principal.anonymous();
     }
 
     /** The caller reconstructed from the transport context. */

fleetd/src/main/java/dev/ltms/fleet/session/SessionManager.java

@@ -302,9 +302,10 @@ public final class SessionManager implements TurnListener {
      * with no copy and no error. Do NOT fuse these back together; the cost of an orphaned worktree
      * is a logged path an operator can reclaim, the cost of a deleted one is unrecoverable work.
      */
-    private void release(String paneId, ReleaseCause cause) {
+    private MemberSession release(String paneId, ReleaseCause cause) {
         MemberSession removed = registry.remove(paneId);
         releaseRemoved(paneId, removed, handles.remove(paneId), cause);
+        return removed;
     }
 
     /**
@@ -1064,16 +1065,39 @@ public final class SessionManager implements TurnListener {
      * drain (see above), and a straggler must not buy the drain more time than the flag it lost the
      * race against would have. In the ordinary case the sweep finds nothing and costs one empty
      * {@link #roster()} call.
+     *
+     * <p>fleetd #512: a drain that releases every session cleanly used to log nothing at all — the
+     * only log calls in this method and {@link #drainSnapshot} sit on abnormal paths, so "nothing
+     * logged" was indistinguishable from "died on the first session". The {@code log.info} at the
+     * end below is a positive assertion that the drain actually finished, on the normal path,
+     * every time — including the all-zero case, which is a common and legitimate outcome (no
+     * members were live) and must still produce the line. Both {@link #drainSnapshot} passes (the
+     * main snapshot and the straggler sweep) are folded into the one line: a caller reading two
+     * lines could not tell a two-pass drain from two separate drains.
      */
     void drainAll(long timeoutNanos) {
         long deadline = System.nanoTime() + timeoutNanos;
         draining.set(true);
-        drainSnapshot(roster(), deadline);
+        DrainTally tally = drainSnapshot(roster(), deadline);
         List<MemberSession> stragglers = roster();
         if (!stragglers.isEmpty()) {
             log.warn("drain sweep found {} session(s) registered after the drain snapshot was "
                     + "taken (raced past the shutdown guard); draining them too", stragglers.size());
-            drainSnapshot(stragglers, deadline);
+            tally = tally.plus(drainSnapshot(stragglers, deadline));
+        }
+        log.info("drain complete: released={} abandoned={} (still BUSY at the shutdown deadline)",
+                tally.released(), tally.abandoned());
+    }
+
+    /**
+     * Running count for one {@link #drainAll} invocation, folded across both {@link #drainSnapshot}
+     * passes (fleetd #512). {@code abandoned} counts sessions that were still {@code BUSY} at the
+     * moment they were released — i.e. the whole-drain deadline passed before they left {@code BUSY}
+     * on their own (see {@link #drainSnapshot}) — a subset of {@code released}, not additional to it.
+     */
+    private record DrainTally(int released, int abandoned) {
+        private DrainTally plus(DrainTally other) {
+            return new DrainTally(released + other.released, abandoned + other.abandoned);
         }
     }
 
@@ -1081,8 +1105,12 @@ public final class SessionManager implements TurnListener {
      * Drain exactly the sessions in {@code snapshot}, waiting out a {@code BUSY} one against the
      * shared whole-drain {@code deadline} before releasing it. Shared by {@link #drainAll}'s main
      * pass and its post-loop straggler sweep (fleetd #308) so both honor the same one budget.
+     * Returns how many sessions this pass released, and how many of those were still {@code BUSY}
+     * (abandoned mid-turn) at the moment of release.
      */
-    private void drainSnapshot(List<MemberSession> snapshot, long deadline) {
+    private DrainTally drainSnapshot(List<MemberSession> snapshot, long deadline) {
+        int released = 0;
+        int abandoned = 0;
         for (MemberSession s : snapshot) {
             try {
                 if (s.state() == MemberSession.State.BUSY) {
@@ -1100,11 +1128,16 @@ public final class SessionManager implements TurnListener {
                         }
                     }
                 }
-                release(s.paneId(), ReleaseCause.SHUTDOWN);
+                MemberSession removed = release(s.paneId(), ReleaseCause.SHUTDOWN);
+                released++;
+                if (removed != null && removed.state() == MemberSession.State.BUSY) {
+                    abandoned++;
+                }
             } catch (RuntimeException e) {
                 log.warn("drain failed for pane={}; continuing with remaining sessions", s.paneId(), e);
             }
         }
+        return new DrainTally(released, abandoned);
     }
 
     /**

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

@@ -186,6 +186,43 @@ class CallerResolverTest {
         assertEquals(Role.PRIMARY, r.resolve("127.0.0.1", 99, "BEARER s3cret").role());
     }
 
+    // ── fleetd #505: a herdr error DURING THE SCAN must not be conflated with "not a worker" ──────
+    // #317 (above) covers a failed lsof lookup. This is the other input to the same decision: the
+    // lsof lookup succeeds (a real pid), but PaneLocator's own pane scan hits a herdr error on the
+    // pane that owns that pid — so c.resolved() is true and c.terminal() is null, exactly like a
+    // real primary. c.scanComplete() is what tells them apart.
+
+    /**
+     * The discriminating case named in the ticket: the error must land on the pane that DOES own
+     * the caller's pid, or the test proves nothing (any other pane's failure is invisible to the
+     * scan's outcome, since a match found elsewhere is definitive regardless).
+     */
+    @Test
+    void aHerdrErrorOnTheOwningPaneDuringTheScanIsRefusedNotPromotedToPrimary() {
+        FakeHerdr failing = new FakeHerdr().processInfoFailsForPane("w2:p7", "transient");
+        ConnectionIdentity incomplete = new ConnectionIdentity(new PaneLocator(failing), _ -> FakeHerdr.WORKER_PID);
+
+        Principal p = new CallerResolver(incomplete).resolve("127.0.0.1", 55555, null);
+
+        assertEquals(Role.ANONYMOUS, p.role(),
+                "an incomplete pane scan must never be read as a clean negative and promoted to primary");
+    }
+
+    /**
+     * The companion invariant: a herdr error on a DIFFERENT, non-owning pane must not turn every
+     * mid-scan teardown into a refusal — the real match is still found and resolves as a worker.
+     */
+    @Test
+    void aHerdrErrorOnANonOwningPaneStillResolvesTheRealWorker() {
+        FakeHerdr vanishedElsewhere = new FakeHerdr().processInfoFailsForPane("w2:p9", "pane_not_found");
+        ConnectionIdentity id = new ConnectionIdentity(new PaneLocator(vanishedElsewhere), _ -> FakeHerdr.WORKER_PID);
+
+        Principal p = new CallerResolver(id).resolve("127.0.0.1", 55555, null);
+
+        assertEquals(Role.WORKER, p.role());
+        assertEquals("term_a", p.terminal());
+    }
+
     @Test
     void aNonLoopbackCallerIsNeverThePrimaryUnderLoopbackTrust() {
         // Defence in depth: startup already refuses this pairing (validateAuthExposure), but if a

fleetd/src/test/java/dev/ltms/fleet/herdr/FakeHerdr.java

@@ -44,6 +44,7 @@ public final class FakeHerdr implements HerdrClient {
     private int workerTabPaneCount = 1;
     private String paneCloseErrorCode = null;
     private final Map<String, String> paneCloseErrorCodeFor = new ConcurrentHashMap<>();
+    private final Map<String, String> processInfoErrorCodeFor = new ConcurrentHashMap<>();
     private String tabCloseErrorCode = null;
     private final Map<String, String> tabCloseErrorCodeFor = new ConcurrentHashMap<>();
     private String agentSendErrorCode = null;
@@ -144,6 +145,18 @@ public final class FakeHerdr implements HerdrClient {
         return this;
     }
 
+    /**
+     * Make {@code pane.process_info} fail with this herdr error code, but only for the given
+     * {@code pane_id} — every other pane's {@code pane.process_info} still succeeds. Models a
+     * transient herdr failure partway through a {@link PaneLocator} pid→pane scan (fleetd #505):
+     * the scan must be able to tell "this pane does not own the pid" apart from "the scan could
+     * not check this pane at all", instead of collapsing both into one {@code false}.
+     */
+    public FakeHerdr processInfoFailsForPane(String paneId, String code) {
+        this.processInfoErrorCodeFor.put(paneId, code);
+        return this;
+    }
+
     /** Set the {@code agent_status} that {@code agent.get} reports (drives the injector). */
     public FakeHerdr agentStatus(String status) {
         this.agentStatus = status;
@@ -407,6 +420,13 @@ public final class FakeHerdr implements HerdrClient {
                           {"pane_id":"w2:p9","terminal_id":"term_shell","workspace_id":"w2","tab_id":"w2:t8"}]}""");
                 case "pane.process_info" -> {
                     Object paneId = params instanceof java.util.Map<?, ?> m ? m.get("pane_id") : null;
+                    String failCode = paneId == null ? null
+                            : processInfoErrorCodeFor.get(String.valueOf(paneId));
+                    if (failCode != null) {
+                        throw new HerdrException(
+                                "herdr error [" + failCode + "]: pane.process_info failed",
+                                failCode, null);
+                    }
                     yield "w2:p7".equals(paneId)
                             ? mapper.readTree(("""
                                 {"type":"pane_process_info","process_info":{"pane_id":"w2:p7","shell_pid":%d,

fleetd/src/test/java/dev/ltms/fleet/herdr/PaneLocatorContractTest.java

@@ -37,7 +37,7 @@ class PaneLocatorContractTest {
                         .path("pane").path("terminal_id").asText(null);
                 assertNotNull(terminalId, "seed pane should carry a terminal_id");
 
-                assertEquals(terminalId, new PaneLocator(herdr).terminalForPid(shellPid),
+                assertEquals(terminalId, new PaneLocator(herdr).terminalForPid(shellPid).terminal(),
                         "a real PID must resolve back to its own pane's terminal_id");
             } finally {
                 spaces.closeTab(tab.tab().tabId());

fleetd/src/test/java/dev/ltms/fleet/herdr/PaneLocatorTest.java

@@ -15,18 +15,20 @@ class PaneLocatorTest {
 
     @Test
     void resolvesTerminalForAForegroundPid() {
-        assertEquals("term_a", loc.terminalForPid(FakeHerdr.WORKER_PID));
+        assertEquals("term_a", loc.terminalForPid(FakeHerdr.WORKER_PID).terminal());
     }
 
     @Test
     void nullForAPidInNoPane() {
-        assertNull(loc.terminalForPid(999_999));
+        PaneLocator.Lookup outcome = loc.terminalForPid(999_999);
+        assertNull(outcome.terminal());
+        assertTrue(outcome.complete(), "a full, error-free scan that finds no match is complete");
     }
 
     @Test
     void nullForNonPositivePid() {
-        assertNull(loc.terminalForPid(0));
-        assertNull(loc.terminalForPid(-1));
+        assertNull(loc.terminalForPid(0).terminal());
+        assertNull(loc.terminalForPid(-1).terminal());
     }
 
     // --- two-daemon fallback (CB-185) -----------------------------------------
@@ -38,7 +40,7 @@ class PaneLocatorTest {
         HerdrClient lead = new FakeHerdr().withNoPanes();
         HerdrClient member = new FakeHerdr();
         PaneLocator two = new PaneLocator(lead, member);
-        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID));
+        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID).terminal());
     }
 
     @Test
@@ -48,13 +50,13 @@ class PaneLocatorTest {
         HerdrClient lead = new FakeHerdr();
         HerdrClient member = new FakeHerdr().withNoPanes();
         PaneLocator two = new PaneLocator(lead, member);
-        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID));
+        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID).terminal());
     }
 
     @Test
     void nullWhenNeitherClientHasTheMatch() {
         PaneLocator two = new PaneLocator(new FakeHerdr().withNoPanes(), new FakeHerdr().withNoPanes());
-        assertNull(two.terminalForPid(FakeHerdr.WORKER_PID));
+        assertNull(two.terminalForPid(FakeHerdr.WORKER_PID).terminal());
     }
 
     @Test
@@ -63,7 +65,7 @@ class PaneLocatorTest {
         // must behave exactly like the one-arg constructor, including making only one herdr call.
         FakeHerdr shared = new FakeHerdr();
         PaneLocator two = new PaneLocator(shared, shared);
-        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID));
+        assertEquals("term_a", two.terminalForPid(FakeHerdr.WORKER_PID).terminal());
         long paneListCalls = shared.calls.stream().filter(c -> c.method().equals("pane.list")).count();
         assertEquals(1, paneListCalls, "same-object lead/member must scan exactly once, not twice");
     }
@@ -75,7 +77,7 @@ class PaneLocatorTest {
         // Regression: a pid with no parent chain at all — no ancestry walk is needed to match it.
         OnePaneHerdr pane = new OnePaneHerdr("term_x", "pX", 5000, 6000);
         PaneLocator loc = new PaneLocator(pane, new FakeParentResolver());
-        assertEquals("term_x", loc.terminalForPid(5000));
+        assertEquals("term_x", loc.terminalForPid(5000).terminal());
     }
 
     @Test
@@ -83,7 +85,7 @@ class PaneLocatorTest {
         // Regression: same as above, but matching via the foreground-processes list.
         OnePaneHerdr pane = new OnePaneHerdr("term_x", "pX", 5000, 6000);
         PaneLocator loc = new PaneLocator(pane, new FakeParentResolver());
-        assertEquals("term_x", loc.terminalForPid(6000));
+        assertEquals("term_x", loc.terminalForPid(6000).terminal());
     }
 
     @Test
@@ -97,7 +99,7 @@ class PaneLocatorTest {
                 .parent(7002, 7001)   // grandchild -> child
                 .parent(7001, 5000);  // child -> shell (the pane's shell_pid)
         PaneLocator loc = new PaneLocator(pane, parents);
-        assertEquals("term_x", loc.terminalForPid(7002));
+        assertEquals("term_x", loc.terminalForPid(7002).terminal());
     }
 
     @Test
@@ -110,7 +112,7 @@ class PaneLocatorTest {
                 .parent(9002, 9001)
                 .parent(9001, 9000); // chain never reaches 5000 or 6000
         PaneLocator loc = new PaneLocator(pane, parents);
-        assertNull(loc.terminalForPid(9002));
+        assertNull(loc.terminalForPid(9002).terminal());
     }
 
     @Test
@@ -122,7 +124,7 @@ class PaneLocatorTest {
                 .parent(100, 101)
                 .parent(101, 100); // cycle, never reaches the pane's pids
         PaneLocator loc = new PaneLocator(pane, parents);
-        assertNull(loc.terminalForPid(100));
+        assertNull(loc.terminalForPid(100).terminal());
     }
 
     @Test
@@ -141,11 +143,69 @@ class PaneLocatorTest {
         };
         HerdrClient noPanes = new FakeHerdr().withNoPanes();
         PaneLocator two = new PaneLocator(noPanes, pane, counting);
-        assertEquals("term_x", two.terminalForPid(7002));
+        assertEquals("term_x", two.terminalForPid(7002).terminal());
         assertEquals(3, calls.get(), "ancestry must be walked once (3 lookups: 7002, 7001, 5000), "
                 + "not re-walked per herdr client");
     }
 
+    // --- fleetd #505: a herdr error during the scan must not read as a clean negative ---------
+
+    @Test
+    void anErrorOnThePaneThatOwnsThePidMakesTheScanIncompleteNotAClearNegative() {
+        // The discriminating case: pane.process_info fails for exactly the pane that DOES own the
+        // caller's pid ("w2:p7", term_a). Before the fix, that failure was swallowed into a plain
+        // "does not own it" and the scan finished with a clean-looking null — indistinguishable
+        // from a real primary. It must now report incomplete, not a definite null.
+        FakeHerdr herdr = new FakeHerdr().processInfoFailsForPane("w2:p7", "transient");
+        PaneLocator loc = new PaneLocator(herdr);
+
+        PaneLocator.Lookup outcome = loc.terminalForPid(FakeHerdr.WORKER_PID);
+
+        assertNull(outcome.terminal(), "the failing pane's ownership could not be confirmed");
+        assertFalse(outcome.complete(),
+                "a scan that could not check the owning pane must not report as complete");
+    }
+
+    @Test
+    void aVanishedPaneThatIsNotTheMatchLeavesAnOtherwiseSuccessfulScanComplete() {
+        // The companion invariant: a DIFFERENT pane (not the caller's own) failing mid-scan must
+        // not turn every mid-scan teardown into a refusal — the real match is still found, and the
+        // scan is still reported complete.
+        FakeHerdr herdr = new FakeHerdr().processInfoFailsForPane("w2:p9", "pane_not_found");
+        PaneLocator loc = new PaneLocator(herdr);
+
+        PaneLocator.Lookup outcome = loc.terminalForPid(FakeHerdr.WORKER_PID);
+
+        assertEquals("term_a", outcome.terminal());
+        assertTrue(outcome.complete(), "a positive match elsewhere in the scan is definitive");
+    }
+
+    // --- fleetd #509: the completeness fold across clients must not collapse to "last wins" ----
+
+    @Test
+    void anEarlierClientsErrorSurvivesALaterClientsCleanNegative() {
+        // terminalForPid folds each client's Lookup.complete() with
+        //     complete = complete && outcome.complete();
+        // (PaneLocator.java:117). With a SINGLE client, a fold that keeps only the last outcome
+        // (dropping the "complete &&" prefix) agrees with the real fold — which is why 14 of the
+        // 15 pre-existing tests never catch that mutation: none of them vary the number of clients.
+        // Here the LEAD client errors on exactly the pane that would have owned the pid (so its
+        // scan is incomplete AND finds no match), and the MEMBER client cleanly reports no panes
+        // at all (a complete, negative scan). The real fold ANDs the two into false. A fold that
+        // just keeps the last client's outcome would read this as a clean true — the earlier
+        // error is erased, and CallerResolver.java:254 would read scanComplete() as true and
+        // promote an unverified caller to the primary.
+        HerdrClient lead = new FakeHerdr().processInfoFailsForPane("w2:p7", "transient");
+        HerdrClient member = new FakeHerdr().withNoPanes();
+        PaneLocator two = new PaneLocator(lead, member);
+
+        PaneLocator.Lookup outcome = two.terminalForPid(FakeHerdr.WORKER_PID);
+
+        assertNull(outcome.terminal(), "the pane that could have owned the pid was never checked");
+        assertFalse(outcome.complete(),
+                "an earlier client's error must survive a later client's clean negative");
+    }
+
     /** Minimal single-pane {@link HerdrClient} fake, purpose-built for the ancestry tests above. */
     private static final class OnePaneHerdr implements HerdrClient {
         private final ObjectMapper mapper = new ObjectMapper();

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

@@ -57,6 +57,23 @@ class ConnectionIdentityTest {
         // must read as "resolved" — the distinction #317 turns on.
         ConnectionIdentity.Caller c = with(_ -> 999_999).resolve("127.0.0.1", 55555);
         assertTrue(c.resolved());
+        assertTrue(c.scanComplete(), "no herdr error happened, so the scan is complete");
+    }
+
+    @Test
+    void scanIsIncompleteWhenHerdrErrorsOnThePaneThatOwnsThePid() {
+        // fleetd #505: a transient herdr error on exactly the pane that DOES own the caller's pid
+        // must be visible as an incomplete scan, distinct from a real primary (resolved(), null
+        // terminal, complete scan). Both have pid > 0 and a null terminal — scanComplete is the
+        // only thing that tells them apart.
+        FakeHerdr failing = new FakeHerdr().processInfoFailsForPane("w2:p7", "transient");
+        ConnectionIdentity id = new ConnectionIdentity(new PaneLocator(failing), _ -> FakeHerdr.WORKER_PID);
+
+        ConnectionIdentity.Caller c = id.resolve("127.0.0.1", 55555);
+
+        assertTrue(c.resolved(), "the pid itself resolved fine — this is not #317's failure");
+        assertNull(c.terminal(), "the owning pane could not be confirmed");
+        assertFalse(c.scanComplete(), "the scan could not check the pane that owns this pid");
     }
 
     @Test

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

@@ -190,6 +190,27 @@ class FleetMcpAuthzTest {
                 "no CallerResolver supplied ⇒ authorization not enforced (legacy behaviour)");
     }
 
+    /**
+     * fleetd #509: {@code legacyPrincipal} (used only when {@code callers == null}, i.e. the
+     * legacy constructor above) used to fall back to {@link Principal#primary} for ANY caller the
+     * connection did not resolve to a worker pane — no {@code isLoopback} check, no
+     * {@code scanComplete} check, unlike the enforced path's {@code CallerResolver.java:254}. A
+     * non-loopback caller (an off-host client) is exactly the case that must never earn the
+     * primary's authority, and authorization being disabled in legacy mode does not make that
+     * safe: the resolved {@link Principal} still reaches non-authz logic such as
+     * {@code markSpawnedMemberPresent} and {@code recordPrimarySingleton}.
+     */
+    @Test
+    void legacyPrincipalIsAnonymousNotPrimaryForAnUnresolvedCaller() {
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> 999_999);
+        // A non-loopback address never even reaches the pane scan — resolve() short-circuits it
+        // to Caller(null, -1, true), the same "no terminal" shape a genuine primary's connection
+        // produces. legacyPrincipal must not conflate the two.
+        Principal p = FleetMcp.legacyPrincipal(identity, "8.8.8.8", 1234);
+        assertEquals(Principal.anonymous(), p,
+                "an unresolved legacy caller must earn no authority, not the primary's");
+    }
+
     // --- fleetd #439: who may see fleet_list's coordinator row ----------------------------------
 
     /**

fleetd/src/test/java/dev/ltms/fleet/session/SessionManagerTest.java

@@ -902,6 +902,104 @@ class SessionManagerTest {
                 .count();
     }
 
+    /**
+     * fleetd #512: a drain that releases every session cleanly used to log nothing at all — the
+     * two log calls in {@code drainAll}/{@code drainSnapshot} both sit on abnormal paths, so
+     * "clean drain" and "died on the first session" were indistinguishable. This asserts the new
+     * {@code log.info} line fires on the ordinary, nothing-went-wrong path, and that its numbers
+     * are the real counts (two released, zero abandoned) rather than just a non-empty string.
+     */
+    @Test
+    void drainAllLogsACompletionLineWithTheRealCountsOnACleanDrain() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession first = sessions.acquire("ltms-local", "/one", "/caller", "ownerOne");
+        MemberSession second = sessions.acquire("ltms-local", "/two", "/caller", "ownerTwo");
+        sessions.asPresence().markPresent(first.terminalId());
+        sessions.asPresence().markPresent(second.terminalId());
+        // Both stay READY — neither is delivered a turn, so neither is BUSY and the drain below
+        // has nothing abnormal to hit.
+
+        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.setContext(ctx);
+        appender.start();
+        sessionLog.addAppender(appender);
+        // Pin INFO explicitly: another test in this class (run order is not guaranteed) leaves the
+        // shared SessionManager logger pinned at WARN via setLevel and never restores it, which
+        // would silently swallow the log.info assertion below.
+        sessionLog.setLevel(Level.INFO);
+        try {
+            sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
+
+            assertTrue(sessions.roster().isEmpty(), "precondition: the drain actually ran");
+            String info = appender.list.stream()
+                    .filter(e -> e.getLevel().equals(Level.INFO))
+                    .map(ILoggingEvent::getFormattedMessage)
+                    .filter(m -> m.contains("drain complete"))
+                    .findFirst()
+                    .orElse("no drain-complete INFO logged");
+            assertTrue(info.contains("released=2"),
+                    "both released sessions must be counted: " + info);
+            assertTrue(info.contains("abandoned=0"),
+                    "neither session was BUSY, so nothing was abandoned mid-turn: " + info);
+        } finally {
+            sessionLog.detachAppender(appender);
+            sessionLog.setLevel(null);
+        }
+    }
+
+    /**
+     * fleetd #512: the same completion line must also report a non-zero abandoned count when a
+     * session is still {@code BUSY} once the whole-drain deadline passes — the case the ticket
+     * calls out as the one a script needs to be able to see. Reuses the same BUSY/READY mix as
+     * {@link #drainAllReleasesBusyAndReadySessionsAndWaitsForBusy}, which already forces the busy
+     * session to spin until the real-time deadline expires (its state never leaves BUSY on its
+     * own), and adds the log assertion that test does not make.
+     */
+    @Test
+    void drainAllLogsANonZeroAbandonedCountForASessionStillBusyAtTheDeadline() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession ready = sessions.acquire("ltms-local", "/ready", "/caller", "ownerR");
+        MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "ownerB");
+        sessions.asPresence().markPresent(ready.terminalId());
+        sessions.asPresence().markPresent(busy.terminalId());
+        sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
+        // busy never leaves BUSY — no completion is delivered — so the drain below must spin the
+        // full timeout and then release it anyway, counting it abandoned.
+
+        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.setContext(ctx);
+        appender.start();
+        sessionLog.addAppender(appender);
+        // Pin INFO explicitly — see the comment in drainAllLogsACompletionLineWithTheRealCountsOnACleanDrain.
+        sessionLog.setLevel(Level.INFO);
+        try {
+            sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
+
+            assertTrue(sessions.roster().isEmpty(), "precondition: the drain actually ran");
+            String info = appender.list.stream()
+                    .filter(e -> e.getLevel().equals(Level.INFO))
+                    .map(ILoggingEvent::getFormattedMessage)
+                    .filter(m -> m.contains("drain complete"))
+                    .findFirst()
+                    .orElse("no drain-complete INFO logged");
+            assertTrue(info.contains("released=2"),
+                    "both the ready and the busy session are released: " + info);
+            assertTrue(info.contains("abandoned=1"),
+                    "the busy session hit the deadline still BUSY and must be counted: " + info);
+        } finally {
+            sessionLog.detachAppender(appender);
+            sessionLog.setLevel(null);
+        }
+    }
+
     // --- fleetd #308: a spawn accepted while the shutdown drain is running must not orphan ---
 
     @Test

scripts/probe-member-credentials.sh

@@ -65,6 +65,27 @@
 # credential the member holds in full.
 #
 set -uo pipefail
+# `pipefail` is not what catches the parser failure handled below (fleetd #500): in
+# `printf '%s' "$POLICY_JSON" | jq -r '...'`, jq is the LAST element of the pipe, so the pipeline's
+# own exit status is already jq's status, with or without pipefail. It is kept as insurance for if
+# a post-processing stage is ever appended after the parser (e.g. `| tail -n +2`) — at that point
+# the parser would sit upstream and pipefail becomes the only thing that still reports its status.
+
+# --- refuse on an interpreter that cannot run this script (fleetd #500) -------------------------
+#
+# mapfile, used below to parse the policy response, was added in bash 4.0. macOS ships bash 3.2.57
+# at /bin/bash, which predates it. This script's own `set -uo pipefail` does not catch a missing
+# mapfile: the builtin just fails with "command not found" on stderr, and every line below that
+# reads the array it would have filled uses a `:-` default or a slice, neither of which `set -u`
+# catches on an unset array. Left unguarded, that chain ends in the "0 known names" refusal further
+# down — a claim about the POLICY, for a failure that is actually about the INTERPRETER. So the
+# interpreter is checked once, explicitly, before it is asked to do anything mapfile depends on.
+if (( ${BASH_VERSINFO[0]} < 4 )); then
+  echo "refusing to run: this script uses mapfile, which needs bash 4 or newer. This shell is bash" \
+       "${BASH_VERSION:-<unknown, no \$BASH_VERSION>}. Re-run it under a newer bash, for example:" \
+       "\"\$(command -v bash)\" \"$0\"" "$@" >&2
+  exit 3
+fi
 
 FLEETD_HOST="${FLEETD_HOST:-http://127.0.0.1:8765}"
 POLICY_URL="${FLEETD_HOST%/}/member-credentials"
@@ -124,16 +145,32 @@ fi
 # One parse pass: line 1 = present (true/false/null), line 2 = policy mode (possibly blank),
 # lines 3-5 = knownCount/allowedCount/blockedCount, remaining lines = the known[] names. A single
 # pass avoids re-parsing (and re-risking a truthiness bug) five separate times.
+#
+# This used to feed the parser straight into `mapfile -t _FIELDS < <(producer)`. That form cannot
+# see the producer fail: `<` `<(...)` is a process substitution, not a pipeline, so `set -o
+# pipefail` does not reach inside it, and mapfile's own exit status reports whether the BUILTIN
+# ran, not whether the command substituted into it succeeded — a failing jq or python3 there still
+# leaves mapfile at rc=0 with an empty array, read as a parse that genuinely found nothing (fleetd
+# #500). Capturing the parser's output with command substitution first, and checking ITS exit
+# status, reports the producer's real failure while the fact still exists — before it is handed to
+# mapfile at all.
+#
+# mapfile then reads from that captured string with `<<<` (a herestring), not `< <(...)`: `<<<`
+# materialises the whole string in memory first, where `< <(...)` would stream it. That only
+# matters for a large producer; this one is a short credential-name policy response, so the
+# tradeoff is irrelevant here — noted because it would not be for every producer.
 if command -v jq >/dev/null 2>&1; then
-  mapfile -t _FIELDS < <(printf '%s' "$POLICY_JSON" | jq -r '
+  _FIELDS_RAW="$(printf '%s' "$POLICY_JSON" | jq -r '
       (.present | tostring),
       (.policy // ""),
       (.knownCount // 0 | tostring),
       (.allowedCount // 0 | tostring),
       (.blockedCount // 0 | tostring),
-      (.known[]? // empty)')
+      (.known[]? // empty)')"
+  _PARSE_STATUS=$?
+  _PARSER_NAME="jq"
 else
-  mapfile -t _FIELDS < <(printf '%s' "$POLICY_JSON" | python3 - <<'PY'
+  _FIELDS_RAW="$(printf '%s' "$POLICY_JSON" | python3 - <<'PY'
 import json, sys
 data = json.load(sys.stdin)
 print(str(data.get("present")))
@@ -144,7 +181,34 @@ print(data.get("blockedCount") if data.get("blockedCount") is not None else 0)
 for n in (data.get("known") or []):
     print(n)
 PY
-  )
+  )"
+  _PARSE_STATUS=$?
+  _PARSER_NAME="python3"
+fi
+
+if [ "$_PARSE_STATUS" -ne 0 ]; then
+  echo "refusing to run: could not parse the policy fetched from $POLICY_URL — $_PARSER_NAME exited" \
+       "non-zero (status $_PARSE_STATUS). That is a parser failure, not a claim about the policy" \
+       "itself; the policy response has not been read." >&2
+  exit 4
+fi
+
+mapfile -t _FIELDS <<< "$_FIELDS_RAW"
+
+# Arity check — the CORRECTNESS fix (fleetd #500). A parser that exits 0 can still return fewer
+# than the 5 fixed fields (present, policy mode, 3 counts) that every line below this expects,
+# whatever the reason: a producer that printed nothing, malformed JSON that jq/python3 still
+# accepted, or a schema change upstream. The slice just below this (`_FIELDS[@]:5`) does not fire
+# `set -u` on an unset OR a short array, and every fixed-field read above used a `:-` default, so
+# without this check a short `_FIELDS` reaches the "0 known names" guard further down with the
+# same look as a policy that genuinely has 0 names. Check the count here, at the one point the
+# fact is still present, before the slice consumes it.
+if (( ${#_FIELDS[@]} < 5 )); then
+  echo "refusing to run: the policy parser ($_PARSER_NAME) returned ${#_FIELDS[@]} field(s); at" \
+       "least 5 are required (present, policy mode, knownCount, allowedCount, blockedCount). The" \
+       "parse ran but its shape is wrong — this is not a claim about how many names the policy" \
+       "knows." >&2
+  exit 5
 fi
 
 PRESENT="${_FIELDS[0]:-null}"
@@ -164,19 +228,21 @@ case "$KNOWN_COUNT_REPORTED" in
     ;;
 esac
 
-# --- guard the denominator explicitly — never proceed on a zero/short count ---------------------
+# --- guard the denominator explicitly — never proceed on a zero count ---------------------------
 #
-# This is the exact trap named in the ticket: an empty (or truncated) NAMES array passes every
-# subsequent "is it set" check vacuously and prints a table that LOOKS complete. So this is checked
-# before anything else runs, with a message that says why, not just that it failed.
+# This is the exact trap named in the ticket: an empty NAMES array passes every subsequent "is it
+# set" check vacuously and prints a table that LOOKS complete. By this point the interpreter gate,
+# the parser-exit-status check, and the arity check above have already ruled out "the interpreter
+# couldn't run mapfile", "the parser failed", and "the parser returned the wrong shape" — so a zero
+# count reaching here really does mean the policy itself reports 0 known names, not a swallowed
+# failure upstream. That is still checked before anything else runs, with a message that says so.
 if [ "${#NAMES[@]}" -eq 0 ] || [ "$KNOWN_COUNT_REPORTED" -eq 0 ]; then
   cat >&2 <<EOF
 refusing to run: the policy fetched from $POLICY_URL contains 0 known names (present=${PRESENT:-unknown}).
 
-Either memberCredentials: is absent/empty on the running daemon (nothing is protected — see fleetd's
-own startup warning), or the response could not be parsed. Either way, checking zero names would
-print a clean-looking table for a policy that protects nothing, or for a probe that read nothing.
-This is refused rather than reported as a pass.
+memberCredentials: is absent or empty on the running daemon — nothing is protected (see fleetd's own
+startup warning). Checking zero names would print a clean-looking table for a policy that protects
+nothing. This is refused rather than reported as a pass.
 EOF
   exit 1
 fi

scripts/redeploy-fleetd.sh

@@ -615,8 +615,9 @@ if [ -n "$OLD_PID" ] && [ "$ASSUME_YES" = 0 ]; then
     # than before this run started, even though the running daemon itself was never touched.
     if [ "$DO_BUILD" = 1 ] && [ -f "$JAR_STAGED" ]; then
       die "aborted — the running daemon was NOT touched, but the freshly built jar is sitting at
-          $JAR_STAGED, not yet swapped into $JAR. Rerun (with or without --no-build) to finish the
-          restart, or remove $JAR_STAGED by hand if you want to discard this build."
+          $JAR_STAGED, not yet swapped into $JAR. Rerun WITHOUT --no-build to finish the restart —
+          the freshly built jar is no longer at the live path that --no-build requires — or
+          remove $JAR_STAGED by hand if you want to discard this build."
     fi
     die "aborted — nothing changed"
   fi

scripts/test-redeploy-fleetd.sh

@@ -206,6 +206,28 @@ test_assert_single_daemon_rejects_two_pids() {
   printf '%s' "$output" | grep -qF '4343' || fail "refusal message does not list the pids it found"
 }
 
+# fleetd #511 — jar_id()'s no-argument default was unpinned by any test: nothing proved it reports
+# $JAR (the live path) rather than $JAR_STAGED. Both halves matter, so this pins both: the bare call
+# must hash the live jar, and an explicit path argument must hash THAT file, not fall back to $JAR.
+# Two files with different content, so a default pointed at the wrong one reports the wrong hash
+# rather than accidentally matching.
+test_jar_id_defaults_to_live_and_reports_explicit_path() {
+  local dir saved_jar="$JAR" saved_staged="$JAR_STAGED"
+  local live_hash staged_hash default_result explicit_result
+  dir="$TMP/jar-id"; mkdir -p "$dir"
+  JAR="$dir/fleetd.jar"; JAR_STAGED="$dir/fleetd-new.jar"
+  printf 'live jar bytes' > "$JAR"
+  printf 'staged jar bytes, not the same content' > "$JAR_STAGED"
+  live_hash="$(shasum -a 256 "$JAR" | cut -c1-12)"
+  staged_hash="$(shasum -a 256 "$JAR_STAGED" | cut -c1-12)"
+  default_result="$(jar_id)"
+  explicit_result="$(jar_id "$JAR_STAGED")"
+  JAR="$saved_jar"; JAR_STAGED="$saved_staged"
+  [ "$live_hash" != "$staged_hash" ] || fail "test fixture error: live and staged jars hashed the same"
+  assert_equals "$live_hash" "$default_result" "jar_id with no arguments must report the hash of \$JAR"
+  assert_equals "$staged_hash" "$explicit_result" "jar_id \"\$JAR_STAGED\" must report the hash of the staged jar, not fall back to \$JAR"
+}
+
 # fleetd #493 — never build into the path a running process holds. stage_built_jar/swap_staged_jar
 # are exercised directly against real files on disk (not stubs), because the whole point is file
 # behavior (does the content move, does the source disappear, does a failure leave both sides
@@ -344,6 +366,26 @@ test_swap_ordered_after_wait_and_before_start() {
     || fail "swap_staged_jar (line $swap_line) is not before the start section (line $start_line)"
 }
 
+# fleetd #511: the drain-gate abort message (fired when a build has staged a jar but the operator
+# declines the drain confirmation) used to tell the operator to "Rerun (with or without --no-build)"
+# to finish the restart. That is wrong — by the time this message can fire, stage_built_jar has
+# already moved the jar off $JAR, so a rerun WITH --no-build hits require_no_build_jar's own refusal
+# ("no jar at $JAR — run without --no-build"). Like test_swap_ordered_after_wait_and_before_start
+# above, this code path is never reached by sourcing (the SOURCED guard stops before the main flow),
+# so the only way to pin its exact wording is to read the source.
+test_drain_gate_abort_message_says_no_no_build() {
+  local src="$ROOT/scripts/redeploy-fleetd.sh" msg
+  msg="$(grep -A3 -F 'aborted — the running daemon was NOT touched, but the freshly built jar is sitting at' "$src")"
+  [ -n "$msg" ] || fail "could not find the drain-gate staged-jar abort message in redeploy-fleetd.sh"
+  if printf '%s' "$msg" | grep -qF 'with or without --no-build'; then
+    fail "abort message still claims a rerun WITH --no-build can finish the restart"
+  fi
+  printf '%s' "$msg" | grep -qF 'WITHOUT --no-build' \
+    || fail "abort message does not tell the operator to rerun without --no-build"
+  printf '%s' "$msg" | grep -qF 'no longer at the live path' \
+    || fail "abort message does not say why --no-build cannot finish the restart"
+}
+
 test_no_errors() {
   cat > "$TMP/no-errors.log" <<'LOG'
 2026-09-05 12:00:00 INFO fleetd listening
@@ -555,6 +597,7 @@ test_require_drivable_supervisor_accepts_known_kinds
 test_count_daemon_pids
 test_assert_single_daemon_accepts_one_pid
 test_assert_single_daemon_rejects_two_pids
+test_jar_id_defaults_to_live_and_reports_explicit_path
 test_stage_built_jar_moves_off_live_path
 test_stage_built_jar_dies_when_build_produced_nothing
 test_swap_staged_jar_moves_staged_onto_live
@@ -565,6 +608,7 @@ test_require_no_build_jar_accepts_present_jar
 test_wait_for_daemon_exit_returns_true_once_pid_clears
 test_wait_for_daemon_exit_times_out_if_pid_never_clears
 test_swap_ordered_after_wait_and_before_start
+test_drain_gate_abort_message_says_no_no_build
 test_no_errors
 test_recovery_patterns_match_source
 test_attributed_recovered_connection_error