charter: separate the blocked forge MCP server from the working GITEA_TOKEN

Two worker reports said they had working forge access, which looked like it
contradicted the charter's "any forge tools it appears to have hold a blocked
credential and fail". Measured: the charter is correct and the reports are
correct. They are about two different credentials.

A worker opens its own PR with curl and a repo-scoped GITEA_TOKEN that the
daemon injects (.claude/skills/implementer/SKILL.md step 5, lines 96-117).
That route works — it is how every worker PR this week was opened. The blocked
credential belongs to the forge MCP server that leaks in from the operator's
user-scope ~/.claude.json, which is a separate thing and does fail every call.

The wording did not distinguish them. A worker reading "any forge tools it
appears to have hold a blocked credential and fail" could reasonably conclude
it cannot reach the forge at all, and skip opening its PR — the one step the
lead depends on. So this is an ambiguity with a cost, not a stale line.

Both sentences now name the MCP server specifically and say plainly that the
injected token is a different, working route.

The canonical block and the wiki template must stay byte-identical. The wiki
submodule is updated in its own tree and the official sync check reports
"in sync: True". The submodule pointer stays unstaged, per the repo rules.

CLAUDE.md

@@ -96,8 +96,10 @@ below are the procedure — run them in order, every task, not only the big ones
    that answers it. **A worker's ask waits ~55 seconds, and no nudge makes that longer** — so never
    brief a worker to "ask me". Decide before you delegate, or give it an explicit default.
 6. **Verify yourself.** Re-run the build and the checks. A worker cannot run your IDE tooling, any
-   forge tools it appears to have hold a blocked credential and fail, and a piped command
-   (`… | tail`) hides failures behind a zero exit — never promote a worker's "clean" to a fact.
+   forge MCP server it appears to have holds a blocked credential and fails every call, and a piped
+   command (`… | tail`) hides failures behind a zero exit — never promote a worker's "clean" to a
+   fact. Its injected repo-scoped `GITEA_TOKEN` is a different credential and does work, so a worker
+   reporting that it opened its own PR is reporting something it really can do.
 7. **Review — fan out.** Spawn reviewers against the diff, one per dimension or per file, with
    `wait:false`. Never the implementer of the scope it reviews, and brief them from the diff — not
    from the implementer's rationale, which carries its own blind spot. Dispatch each PR's reviewers
@@ -200,8 +202,11 @@ simply complies has thrown away the reason there are two of you.
    assume them.** What you mount depends on your backend: an opencode member gets the bridge and
    nothing else, while a Claude Code member also inherits the operator's user-scope MCP servers,
    which the bridge never chose for you. Two rules follow. The primary's IDE tooling is still not
-   yours, whatever you see. And **a mounted tool is not a working tool** — the forge server you may
-   find there holds a deliberately blocked credential and fails every call, by design.
+   yours, whatever you see. And **a mounted tool is not a working tool** — the forge MCP server you
+   may find there holds a deliberately blocked credential and fails every call, by design. That is
+   not your only forge route, and the two must not be confused: the repo-scoped `GITEA_TOKEN` the
+   daemon injects into your environment does work, and using it to open your own PR is part of the
+   job. A blocked MCP tool is never a reason to skip that step.
 6. **Never merge.** Stage files explicitly — never `git add -A` — and leave alone anything the
    project marks as not-yours-to-commit.
 

fleetd/src/main/java/dev/ltms/fleet/Fleetd.java

@@ -694,7 +694,7 @@ public final class Fleetd {
         }, outagePolicy);
 
         FleetMcp mcp = new FleetMcp(messages, workers, sessions, identity, presence,
-                primaryRegistry, callers, metrics,
+                primaryRegistry, callers, FleetMcp.AuthorizationMode.ENFORCED, metrics,
                 capacitySource(config, cfg, profile -> liveCountRef.get().apply(profile)),
                 new FleetMcp.HealthCoverageSource(() -> {
                     var health = config.get().health();

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

@@ -93,7 +93,18 @@ public final class FleetMcp {
 
     private final HttpServletStreamableServerTransportProvider transport;
     private final McpSyncServer server;
-    private final CallerResolver authz; // CB-501: null → authorization not enforced (legacy)
+    /**
+     * fleetd #518: whether {@link #denyFor} enforces the CB-505 policy table at all. Replaces the
+     * old {@code CallerResolver authz} field, whose null-ness used to decide BOTH this AND which
+     * principal-resolution code path {@link #contextExtractor} ran — reaching "authorization off"
+     * by simply not passing a {@link CallerResolver} also meant the resolved {@link Principal}
+     * came from a second, separately-maintained heuristic ({@code legacyPrincipal}, now deleted)
+     * that nothing ever exercised. There is now exactly one resolution path ({@code callers},
+     * required and non-null below) and a separate, explicitly-chosen {@link AuthorizationMode}
+     * for this flag — so a caller can turn enforcement off without silently swapping in a second,
+     * untested identity heuristic.
+     */
+    private final boolean authorizationEnforced;
     private final Metrics metrics;      // CB-502: null → auth failures not counted
     private final CapacitySource capacity;
     private final HealthCoverageSource healthCoverage;
@@ -250,6 +261,17 @@ public final class FleetMcp {
         public static CoordinationSource none() { return new CoordinationSource(null, List.of()); }
     }
 
+    /**
+     * fleetd #518: whether {@link #denyFor} enforces the CB-505 policy table. A required
+     * constructor parameter with no default, so "authorization is off" can only be reached by a
+     * caller explicitly saying so — never by omitting a {@link CallerResolver} the way the old
+     * {@code callers == null} idiom allowed. {@code callers} itself is required either way: even
+     * under {@link #UNENFORCED}, the one real {@link CallerResolver} still resolves every caller's
+     * {@link Principal} (so {@code markSpawnedMemberPresent}/{@code recordPrimarySingleton} see a
+     * real identity), and {@link #denyFor} is the only thing that changes.
+     */
+    public enum AuthorizationMode { ENFORCED, UNENFORCED }
+
     /**
      * The only constructor (fleetd #480 Unit C correction round). Every field below used to have
      * its own defaulting overload — {@code leadChannel}/{@code outage}/{@code leadSeats}/
@@ -268,10 +290,16 @@ public final class FleetMcp {
      * {@link OutageSource#none()}, {@link LeadSeatSource#none()}, {@code List.of()} are all still
      * perfectly fine values, just never an implicit default reached by omission.
      *
-     * @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 Jetty's context handler and never passes through
-     *                     Javalin's {@code before} filter, so the REST guard does not cover it.
+     * @param callers      resolves each call's {@link Principal}. Required, never {@code null} —
+     *                     fleetd #518: use {@link AuthorizationMode#UNENFORCED} to disable
+     *                     enforcement, not a missing resolver. This surface needs its own
+     *                     enforcement: {@code /mcp} is a raw servlet on Jetty's context handler and
+     *                     never passes through Javalin's {@code before} filter, so the REST guard
+     *                     does not cover it.
+     * @param authorizationMode fleetd #518: whether {@link #denyFor} enforces the CB-505 policy
+     *                     table ({@link AuthorizationMode#ENFORCED}) or leaves the gate open
+     *                     ({@link AuthorizationMode#UNENFORCED}, for the pre-CB-513 test suite that
+     *                     does not exercise authorization). Required, with no default.
      * @param metrics      registry for auth-failure counting; may be {@code null}
      * @param quarantine   CB-578 stage B facts for {@code fleet_profiles}; pass
      *                     {@link QuarantineSource#none()} for a caller that does not want the
@@ -301,9 +329,13 @@ public final class FleetMcp {
      */
     public FleetMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
                      ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
-                      CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage,
+                      CallerResolver callers, AuthorizationMode authorizationMode, Metrics metrics,
+                      CapacitySource capacity, HealthCoverageSource healthCoverage,
                       QuarantineSource quarantine, LeadChannel leadChannel, OutageSource outage,
                       LeadSeatSource leadSeats, List<String> peers, LeadRollover leadRollover) {
+        Objects.requireNonNull(callers, "callers");
+        this.authorizationEnforced = Objects.requireNonNull(authorizationMode, "authorizationMode")
+                == AuthorizationMode.ENFORCED;
         this.leadChannel = leadChannel;
         this.peers = peers == null ? List.of() : List.copyOf(peers);
         this.capacity = capacity;
@@ -322,11 +354,11 @@ public final class FleetMcp {
                 // (CB-113) — its MCP initialize is the reliable "the agent is up" signal.
                 .contextExtractor(req -> {
                     // One resolution per call, shared with the REST surface via CallerResolver so
-                    // the two paths cannot drift on who a caller is.
-                    Principal p = callers != null
-                            ? callers.resolve(req.getRemoteAddr(), req.getRemotePort(),
-                                    req.getHeader("Authorization"))
-                            : legacyPrincipal(identity, req.getRemoteAddr(), req.getRemotePort());
+                    // the two paths cannot drift on who a caller is. fleetd #518: callers is
+                    // required (never null) so there is no second, untested resolution path to
+                    // fall back to here — AuthorizationMode governs enforcement, not identity.
+                    Principal p = callers.resolve(req.getRemoteAddr(), req.getRemotePort(),
+                            req.getHeader("Authorization"));
                     // CB-532: guard on the ROLE, not on the terminal being null. This excludes a
                     // lead, which carries its pane too, while including every spawned member role.
                     // Enrolling a lead would count it as an available member in the roster.
@@ -443,7 +475,7 @@ public final class FleetMcp {
                     McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_list", Map.of()), null);
                     if (denied != null) return denied;
                     return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, outage,
-                            leadSeats, callers == null ? Map.of() : callers.leads(),
+                            leadSeats, callers.leads(),
                             callerTerminal(exchange),
                             new CoordinationSource(leadChannel, peers),
                             coordinatorVisibleTo(principal(exchange)));
@@ -522,21 +554,9 @@ public final class FleetMcp {
                 .toolCall(fleetWhoami, whoamiHandler)
                 .toolCall(fleetHandover, handoverHandler)
                 .build();
-        this.authz = callers;
         this.metrics = metrics;
     }
 
-    /**
-     * 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.
-     */
-    private 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());
-    }
-
     /** The caller reconstructed from the transport context. */
     private static Principal principal(McpSyncServerExchange exchange) {
         return principalFrom(exchange.transportContext().get(CALLER_ROLE),
@@ -591,8 +611,8 @@ public final class FleetMcp {
     McpSchema.CallToolResult denyFor(Principal caller, Authz.Action action, String target) {
         // The enforcement switch lives HERE rather than in the exchange-facing wrapper: any future
         // tool that calls this directly must not be able to skip the gate by accident.
-        if (authz == null) {
-            return null; // legacy constructor: authorization not enforced
+        if (!authorizationEnforced) {
+            return null; // AuthorizationMode.UNENFORCED: authorization not enforced (fleetd #518)
         }
         if (Authz.permits(caller, action, target)) {
             if (action != Authz.Action.READ) {

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

@@ -78,10 +78,15 @@ class FleetMcpAuthzTest {
         // fleetd #480 correction round: FleetMcp has one constructor now (no defaulting
         // overloads — see its javadoc), so every feature this test does not exercise is passed
         // its explicit "off" value here rather than being omitted.
+        //
+        // fleetd #518: callers is now required (never null) either way — the resolver that used
+        // to be omitted to reach "legacy" is now always real, and AuthorizationMode is the
+        // separate, explicit choice that governs enforcement.
         mcp = new FleetMcp(messages, workers, sessions, identity, sessions.asPresence(),
                 new PrimaryRegistry(null),
-                enforce ? CallerResolver.withLeadsAndMembers(identity, false, null,
-                Map::of, new MemberRegistry(null)) : null,
+                CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, new MemberRegistry(null)),
+                enforce ? FleetMcp.AuthorizationMode.ENFORCED : FleetMcp.AuthorizationMode.UNENFORCED,
                 metrics, FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), null, FleetMcp.OutageSource.none(),
                 FleetMcp.LeadSeatSource.none(), List.of(), null);
@@ -187,7 +192,31 @@ class FleetMcpAuthzTest {
         // The 22 pre-existing FleetMcpTest cases rely on no authorization being enforced.
         FleetMcp m = mcp(false);
         assertNull(m.denyFor(ANON, Authz.Action.SPAWN, null),
-                "no CallerResolver supplied ⇒ authorization not enforced (legacy behaviour)");
+                "AuthorizationMode.UNENFORCED chosen explicitly ⇒ authorization not enforced "
+                        + "(legacy behaviour) — fleetd #518 replaced the old callers == null idiom");
+    }
+
+    /**
+     * fleetd #509 was originally proven against {@code FleetMcp.legacyPrincipal} — a second,
+     * separately-maintained principal-resolution heuristic that only ran when {@code callers} was
+     * omitted (null). fleetd #518 deleted that whole heuristic: {@code callers} is now required
+     * and non-null under every {@link FleetMcp.AuthorizationMode}, so the ONE real
+     * {@link CallerResolver} resolves every caller, enforced or not, and #509's property (a
+     * non-loopback / unresolved caller must never earn the primary's authority) is exactly what
+     * {@code CallerResolverTest.aNonLoopbackCallerIsNeverThePrimaryUnderLoopbackTrust} already
+     * proves on that one real path. There is no longer a second heuristic here to test.
+     */
+    @Test
+    void anUnresolvedNonLoopbackCallerIsAnonymousUnderTheOneRealResolver() {
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> 999_999);
+        CallerResolver resolver = CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, new MemberRegistry(null));
+        // 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 on loopback. The real resolver must not conflate the two.
+        Principal p = resolver.resolve("8.8.8.8", 1234, null);
+        assertEquals(Principal.anonymous(), p,
+                "an unresolved, non-loopback 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/mcp/FleetMcpContextExtractorTest.java

@@ -0,0 +1,147 @@
+package dev.ltms.fleet.mcp;
+
+import dev.ltms.fleet.auth.CallerResolver;
+import dev.ltms.fleet.auth.MemberRegistry;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
+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.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.msg.InMemoryReplyInbox;
+import dev.ltms.fleet.msg.MessageService;
+import dev.ltms.fleet.msg.Rendezvous;
+import dev.ltms.fleet.session.FakeWorktrees;
+import dev.ltms.fleet.session.SessionManager;
+import io.modelcontextprotocol.client.McpClient;
+import io.modelcontextprotocol.client.McpSyncClient;
+import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.http.HttpRequest;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #518 — Part 2: drive the {@code contextExtractor} closure for real.
+ *
+ * <p>{@code FleetMcp.deny()}/{@code denyFor()} has a full policy table of tests
+ * ({@code FleetMcpAuthzTest}), and {@code CallerResolver.resolve()} has its own full suite
+ * ({@code CallerResolverTest}). Neither one ever exercises the closure that WIRES them together
+ * inside {@code FleetMcp}'s constructor: it is built once, handed to the MCP SDK's transport, and
+ * only ever runs when a real MCP client makes a real HTTP request. Every existing test either
+ * calls {@code denyFor(Principal, ...)} with a hand-built {@link dev.ltms.fleet.auth.Principal}
+ * (never asking who the transport would actually have resolved) or drives a static handler method
+ * directly. A mutation that swapped the whole resolution decision for an unconditional fallback —
+ * bypassing {@link CallerResolver} entirely — passed the full suite, including every
+ * {@code FleetMcpAuthzTest} case, because none of them go through the transport at all.
+ *
+ * <p>This test boots the real {@code HttpServletStreamableServerTransportProvider} on a real
+ * Jetty server, drives it with a real MCP client over HTTP, and checks a result that only the
+ * real {@link CallerResolver} can produce: token-mode inspects the {@code Authorization} header
+ * and grants {@code PRIMARY} only for the right bearer token. The connection never resolves to a
+ * worker pane (the fake peer-pid lookup always misses), so the ONLY way {@code fleet_whoami} can
+ * come back as {@code primary} is if the closure actually called {@code callers.resolve(...)} and
+ * read that header — a behaviour the deleted {@code legacyPrincipal} heuristic never had at all.
+ */
+class FleetMcpContextExtractorTest {
+
+    private static final String TOKEN = "s3cret-mcp-token";
+
+    private final FakeHerdr herdr = new FakeHerdr();
+    private final AgentControl agents = new AgentControl(herdr);
+    private FleetMcp mcp;
+    private Server server;
+
+    @AfterEach
+    void tearDown() throws Exception {
+        if (server != null) {
+            server.stop();
+        }
+        if (mcp != null) {
+            mcp.close();
+        }
+    }
+
+    @Test
+    void aRealMcpRequestIsResolvedByTheRealCallerResolverNotAFallback() throws Exception {
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(agents, new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
+                _ -> "tok");
+        SessionManager sessions = new SessionManager(workers, new FakeWorktrees());
+        MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous(),
+                new InMemoryReplyInbox());
+        // The peer-pid lookup always misses (-1), so no connection here is ever resolved to a
+        // worker pane — every call falls through to CallerResolver's token check, the one branch
+        // that is unreachable through the deleted legacy heuristic.
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> -1);
+        CallerResolver callers = CallerResolver.withLeadsAndMembers(identity, true, TOKEN,
+                Map::of, new MemberRegistry(null));
+
+        mcp = new FleetMcp(messages, workers, sessions, identity, sessions.asPresence(),
+                new PrimaryRegistry(null), callers, FleetMcp.AuthorizationMode.ENFORCED,
+                null, FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), null, FleetMcp.OutageSource.none(),
+                FleetMcp.LeadSeatSource.none(), List.of(), null);
+
+        ServletContextHandler handler = new ServletContextHandler();
+        handler.setContextPath("/");
+        handler.addServlet(new ServletHolder(mcp.servlet()), "/mcp");
+        server = new Server(0);
+        server.setHandler(handler);
+        server.start();
+        String baseUrl = "http://127.0.0.1:"
+                + ((ServerConnector) server.getConnectors()[0]).getLocalPort();
+
+        // The right bearer token: the real CallerResolver grants PRIMARY, which fleet_whoami's
+        // READ gate lets through.
+        McpSchema.CallToolResult authorized = callWhoami(baseUrl, "Bearer " + TOKEN);
+        assertFalse(authorized.isError(), "a valid bearer token must resolve as PRIMARY and pass "
+                + "fleet_whoami's READ gate: " + textOf(authorized));
+        assertTrue(textOf(authorized).contains("\"role\":\"primary\""),
+                "fleet_whoami must report the role the real CallerResolver resolved over this "
+                        + "connection, not a fallback: " + textOf(authorized));
+
+        // No credential at all, over the SAME wiring: the real resolver refuses it as ANONYMOUS.
+        // legacyPrincipal never looked at the Authorization header, so it could not have told
+        // these two calls apart at all -- this is the assertion the deleted mutation would fail.
+        McpSchema.CallToolResult unauthorized = callWhoami(baseUrl, null);
+        assertTrue(unauthorized.isError(), "no credential must be refused, not silently let "
+                + "through: " + textOf(unauthorized));
+    }
+
+    private static McpSchema.CallToolResult callWhoami(String baseUrl, String authorizationHeader) {
+        HttpRequest.Builder requestTemplate = HttpRequest.newBuilder();
+        if (authorizationHeader != null) {
+            requestTemplate.header("Authorization", authorizationHeader);
+        }
+        McpClientTransport transport = HttpClientStreamableHttpTransport.builder(baseUrl)
+                .endpoint("/mcp")
+                .requestBuilder(requestTemplate)
+                .build();
+        try (McpSyncClient client = McpClient.sync(transport).build()) {
+            client.initialize();
+            return client.callTool(McpSchema.CallToolRequest.builder("fleet_whoami").arguments(Map.of()).build());
+        }
+    }
+
+    private static String textOf(McpSchema.CallToolResult r) {
+        return ((McpSchema.TextContent) r.content().getFirst()).text();
+    }
+}

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

@@ -89,6 +89,7 @@ class FleetMcpHandoverTest {
         mcp = new FleetMcp(messages, workers, sessions, identity, sessions.asPresence(),
                 new PrimaryRegistry(null),
                 CallerResolver.withLeadsAndMembers(identity, false, null, Map::of, new MemberRegistry(null)),
+                FleetMcp.AuthorizationMode.ENFORCED,
                 null, FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), null, FleetMcp.OutageSource.none(),
                 FleetMcp.LeadSeatSource.none(), List.of(), leadRollover);

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

@@ -25,7 +25,12 @@ import dev.ltms.fleet.peer.SpawnRequest;
 import dev.ltms.fleet.placement.BackendQuarantine;
 import dev.ltms.fleet.placement.PlacementDecision;
 import dev.ltms.fleet.placement.PlacementPolicies;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
 import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.LoggerFactory;
 
@@ -50,9 +55,46 @@ import static org.junit.jupiter.api.Assertions.*;
  * CB-301 / CB-303 acceptance tests for the authoritative session registry, one-shot lifecycle FSM,
  * and configurable lifecycle limits (idle TTL, context cap, drain).
  * No live herdr — everything runs against the same {@link FakeHerdr} the rest of the project uses.
+ *
+ * <p>fleetd #525: only {@link #onTurnFailedIsLoggedAtWarnWithThePriorState} (explicitly
+ * {@link Order#value() @Order(1)}) and the proving test right after it
+ * ({@link #sharedSessionManagerLoggerLevelIsRestoredAfterOnTurnFailedPinsWarn}, {@code @Order(2)})
+ * care about method order — every other test here has no {@code @Order} and so runs after both of
+ * these (JUnit 5's {@link MethodOrderer.OrderAnnotation} gives an unannotated method the lowest
+ * priority), in whatever relative order it already ran in.
  */
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 class SessionManagerTest {
 
+    /**
+     * fleetd #525: the level {@link SessionManager}'s logger had when this class started, captured
+     * before any test here — including the leak this ticket fixes — can touch it. {@code
+     * pinSessionManagerLoggerToAKnownBaseline} then forces a distinctive, known value (DEBUG) so
+     * {@link #sharedSessionManagerLoggerLevelIsRestoredAfterOnTurnFailedPinsWarn} can tell "the
+     * level came back to what it was" apart from "the level happens to already be WARN because
+     * some earlier test class in this JVM fork (surefire reuses forks by default) left it there" —
+     * a real risk, since {@code ch.qos.logback.classic.Logger} instances are cached per class and
+     * shared across the whole JVM, and this exact logger is also touched by
+     * {@code WorktreeSessionManagerTest#releasePreservesDirtyWorktreeAndLogsWarn}, which has the
+     * same unfixed leak (reported, not fixed — out of this ticket's scope).
+     */
+    private static Level sessionManagerLevelBeforeThisClass;
+
+    @BeforeAll
+    static void pinSessionManagerLoggerToAKnownBaseline() {
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        sessionManagerLevelBeforeThisClass = sessionLog.getLevel();
+        sessionLog.setLevel(Level.DEBUG);
+    }
+
+    @AfterAll
+    static void restoreSessionManagerLoggerLevel() {
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        sessionLog.setLevel(sessionManagerLevelBeforeThisClass);
+    }
+
     private SessionManager sessionManager(FakeHerdr herdr) {
         FleetConfig.Profile cfg = new FleetConfig.Profile(
                 "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
@@ -81,6 +123,55 @@ class SessionManagerTest {
         return new SessionManager(workers, worktrees, clock);
     }
 
+    /**
+     * fleetd #525: captures a logger's output and, on {@link #close}, restores <em>both</em> the
+     * appender and the level to what they were before. A bare {@code addAppender}/{@code
+     * setLevel} pair whose {@code finally} only detaches the appender leaves the level pinned —
+     * {@code ch.qos.logback.classic.Logger} instances are cached per class and shared across the
+     * whole JVM, so a level set by one test in this class is still in effect for every test that
+     * runs after it, in this class or any other. try-with-resources makes "restored the appender
+     * but not the level" impossible to write, because there is only one thing to close.
+     */
+    private static final class CapturedLog implements AutoCloseable {
+        private final ch.qos.logback.classic.Logger logger;
+        private final Level originalLevel;
+        private final ListAppender<ILoggingEvent> appender;
+
+        private CapturedLog(Class<?> loggerClass, Level pinnedLevel) {
+            LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+            this.logger = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(loggerClass);
+            this.originalLevel = logger.getLevel();
+            this.appender = new ListAppender<>();
+            appender.setContext(ctx);
+            appender.start();
+            logger.addAppender(appender);
+            if (pinnedLevel != null) {
+                logger.setLevel(pinnedLevel);
+            }
+        }
+
+        /** Capture {@code loggerClass}'s output, pinning its level to {@code pinnedLevel} for the
+         *  duration of the try-with-resources block. */
+        static CapturedLog at(Class<?> loggerClass, Level pinnedLevel) {
+            return new CapturedLog(loggerClass, pinnedLevel);
+        }
+
+        /** Capture {@code loggerClass}'s output without changing its level. */
+        static CapturedLog of(Class<?> loggerClass) {
+            return new CapturedLog(loggerClass, null);
+        }
+
+        List<ILoggingEvent> events() {
+            return appender.list;
+        }
+
+        @Override
+        public void close() {
+            logger.detachAppender(appender);
+            logger.setLevel(originalLevel);
+        }
+    }
+
     /**
      * CB-581: a {@link Worktrees} test double whose {@code hasUncommitted} and {@code remove} can
      * be told to throw, so {@link SessionManager#release} can be exercised against exactly the
@@ -466,24 +557,15 @@ class SessionManagerTest {
 
     @Test
     void backendErrorForUnknownTargetIsWarnedAndDoesNotCreateASession() {
-        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);
-        try {
+        try (CapturedLog log = CapturedLog.of(SessionManager.class)) {
             SessionManager sessions = sessionManager(new FakeHerdr());
 
             assertFalse(sessions.onBackendError("term_missing", "backend exited"));
 
             assertTrue(sessions.roster().isEmpty(), "unknown target must not create a session");
-            assertTrue(appender.list.stream().anyMatch(e -> e.getLevel().equals(Level.WARN)
+            assertTrue(log.events().stream().anyMatch(e -> e.getLevel().equals(Level.WARN)
                             && e.getFormattedMessage().contains("term_missing")),
                     "unknown target is logged at WARN");
-        } finally {
-            sessionLog.detachAppender(appender);
         }
     }
 
@@ -516,19 +598,12 @@ class SessionManagerTest {
     }
 
     @Test
+    @Order(1)
     void onTurnFailedIsLoggedAtWarnWithThePriorState() {
         // CB-564: this transition used to be a bare DEBUG "session marked failed" — a symptom with no
         // cause. A member that can no longer be delegated to must be at least WARN, and should name
         // what stage it failed at (here: BUSY, i.e. a turn was in flight and never resolved).
-        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);
-        sessionLog.setLevel(Level.WARN);
-        try {
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.WARN)) {
             FakeHerdr herdr = new FakeHerdr();
             SessionManager sessions = sessionManager(herdr);
             MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
@@ -538,18 +613,37 @@ class SessionManagerTest {
 
             sessions.onTurnFailed(terminal);
 
-            String warn = appender.list.stream()
+            String warn = log.events().stream()
                     .filter(e -> e.getLevel().equals(Level.WARN))
                     .map(ILoggingEvent::getFormattedMessage)
                     .findFirst()
                     .orElse("no turn-failed WARN logged");
             assertTrue(warn.contains(terminal), "the log names the member: " + warn);
             assertTrue(warn.contains("BUSY"), "the log names the stage it failed at: " + warn);
-        } finally {
-            sessionLog.detachAppender(appender);
         }
     }
 
+    /**
+     * fleetd #525: proves the leak in {@link #onTurnFailedIsLoggedAtWarnWithThePriorState} above
+     * (which runs immediately before this, via {@code @Order}) is closed. That test pins the
+     * shared {@link SessionManager} logger to WARN through a {@link CapturedLog}; if {@link
+     * CapturedLog#close} only detached the appender — the original bug, before this ticket's fix —
+     * the level would still read WARN here instead of the {@code DEBUG} baseline this class's
+     * {@code @BeforeAll} set. Runs at {@code @Order(2)}, guaranteed after {@code @Order(1)} and
+     * before every other (unannotated) test in this class.
+     */
+    @Test
+    @Order(2)
+    void sharedSessionManagerLoggerLevelIsRestoredAfterOnTurnFailedPinsWarn() {
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        assertEquals(Level.DEBUG, sessionLog.getLevel(),
+                "onTurnFailedIsLoggedAtWarnWithThePriorState pins the shared SessionManager logger "
+                        + "to WARN; its cleanup must restore the level it captured (DEBUG, set by "
+                        + "this class's @BeforeAll) rather than leaving WARN pinned for every test "
+                        + "that runs after it");
+    }
+
     /**
      * fleetd #226: a contended slot is refused through the real {@link SessionManager#acquire}
      * path before the real launcher can hand an architect charter to a process.
@@ -576,28 +670,18 @@ class SessionManagerTest {
         SessionManager sessions = sessionManager(herdr);
         MemberRegistry members = architectRegistry();
         sessions.setMemberLifecycle(bindFailureAfterReservation(members));
-        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
-        ch.qos.logback.classic.Logger registryLog = (ch.qos.logback.classic.Logger)
-                LoggerFactory.getLogger(MemberRegistry.class);
-        ListAppender<ILoggingEvent> appender = new ListAppender<>();
-        appender.setContext(ctx);
-        appender.start();
-        registryLog.addAppender(appender);
-        registryLog.setLevel(Level.WARN);
-        try {
+        try (CapturedLog log = CapturedLog.at(MemberRegistry.class, Level.WARN)) {
             MemberSession session = sessions.acquire("ltms-local", MemberRole.ARCHITECT, null,
                     "/caller", "term_primary", null);
 
             assertEquals(MemberRole.DEV, session.role(), "a failed reservation bind must use the fallback");
-            String warn = appender.list.stream()
+            String warn = log.events().stream()
                     .filter(e -> e.getLevel().equals(Level.WARN))
                     .map(ILoggingEvent::getFormattedMessage)
                     .findFirst()
                     .orElse("no slot-exhaustion WARN logged");
             assertTrue(warn.contains("ltms-local"), "the WARN names the profile: " + warn);
             assertTrue(warn.contains(session.terminalId()), "the WARN names the terminal: " + warn);
-        } finally {
-            registryLog.detachAppender(appender);
         }
     }
 
@@ -902,6 +986,82 @@ 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.
+
+        // Pin INFO explicitly: fleetd #525 made CapturedLog itself restore the level it pins, but
+        // this pin stays anyway as belt-and-braces — a later change to the sweep must not be able
+        // to make this INFO assertion vacuous again by leaving some other test's WARN pin in place.
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.INFO)) {
+            sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
+
+            assertTrue(sessions.roster().isEmpty(), "precondition: the drain actually ran");
+            String info = log.events().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);
+        }
+    }
+
+    /**
+     * 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.
+
+        // Pin INFO explicitly — see the comment in drainAllLogsACompletionLineWithTheRealCountsOnACleanDrain.
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.INFO)) {
+            sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
+
+            assertTrue(sessions.roster().isEmpty(), "precondition: the drain actually ran");
+            String info = log.events().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);
+        }
+    }
+
     // --- fleetd #308: a spawn accepted while the shutdown drain is running must not orphan ---
 
     @Test
@@ -1202,21 +1362,13 @@ class SessionManagerTest {
                 new WorktreeRequest("cb-581a", null));
         worktrees.failHasUncommittedWith(new WorktreeException("git status exited 128"));
 
-        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);
-        sessionLog.setLevel(Level.WARN);
-        try {
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.WARN)) {
             assertDoesNotThrow(() -> sessions.release(s.paneId()),
                     "a throwing dirty check must not abort the release");
 
             assertTrue(worktrees.removeCalls().isEmpty(),
                     "the worktree is preserved when its dirty state cannot be determined");
-            String warn = appender.list.stream()
+            String warn = log.events().stream()
                     .filter(e -> e.getLevel().equals(Level.WARN))
                     .map(ILoggingEvent::getFormattedMessage)
                     .filter(m -> m.contains(s.worktree()))
@@ -1224,8 +1376,6 @@ class SessionManagerTest {
                     .orElse("no warn logged naming the worktree");
             assertTrue(warn.contains(s.paneId()), "the WARN names the pane: " + warn);
             assertTrue(warn.contains(s.terminalId()), "the WARN names the terminal: " + warn);
-        } finally {
-            sessionLog.detachAppender(appender);
         }
     }
 
@@ -1392,20 +1542,12 @@ class SessionManagerTest {
         // this itself, so it no longer propagates out of release() at all.
         worktrees.failRemoveFor(b.worktree());
 
-        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);
-        sessionLog.setLevel(Level.WARN);
         int reaped;
-        try {
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.WARN)) {
             clock[0] = 100;
             reaped = sessions.reapIdle(10);
 
-            String warn = appender.list.stream()
+            String warn = log.events().stream()
                     .filter(e -> e.getLevel().equals(Level.WARN))
                     .map(ILoggingEvent::getFormattedMessage)
                     .filter(m -> m.contains(b.paneId()))
@@ -1413,8 +1555,6 @@ class SessionManagerTest {
                     .orElse("no worktree-removal-failure WARN logged");
             assertTrue(warn.contains(b.terminalId()), "the WARN names the failed session's terminal: " + warn);
             assertTrue(warn.contains(b.worktree()), "the WARN names the failed session's worktree: " + warn);
-        } finally {
-            sessionLog.detachAppender(appender);
         }
 
         assertEquals(3, reaped,
@@ -1460,28 +1600,18 @@ class SessionManagerTest {
         // trigger reapIdle's own guard is for, now that #283 closed the worktree-removal trigger.
         herdr.paneCloseFailsForPane("w9:pRoot_2", "internal_error");
 
-        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);
-        sessionLog.setLevel(Level.WARN);
         int reaped;
-        try {
+        try (CapturedLog log = CapturedLog.at(SessionManager.class, Level.WARN)) {
             clock[0] = 100;
             reaped = sessions.reapIdle(10);
 
-            String warn = appender.list.stream()
+            String warn = log.events().stream()
                     .filter(e -> e.getLevel().equals(Level.WARN))
                     .map(ILoggingEvent::getFormattedMessage)
                     .filter(m -> m.contains("reap failed") && m.contains(b.paneId()))
                     .findFirst()
                     .orElse("no reap-failed WARN logged for the failing session");
             assertTrue(warn.contains(b.terminalId()), "the WARN names the failed session's terminal: " + warn);
-        } finally {
-            sessionLog.detachAppender(appender);
         }
 
         assertEquals(2, reaped,

scripts/probe-member-credentials.sh

@@ -64,7 +64,106 @@
 # The two outputs side by side are the finding: any name whose hash matches between them is a
 # credential the member holds in full.
 #
+# 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.
+parse_policy_fields() {
+  if command -v jq >/dev/null 2>&1; then
+    _FIELDS_RAW="$(printf '%s' "$POLICY_JSON" | jq -r '
+        (.present | tostring),
+        (.policy // ""),
+        (.knownCount // 0 | tostring),
+        (.allowedCount // 0 | tostring),
+        (.blockedCount // 0 | tostring),
+        (.known[]? // empty)')"
+    _PARSE_STATUS=$?
+    _PARSER_NAME="jq"
+  else
+    _FIELDS_RAW="$(printf '%s' "$POLICY_JSON" | python3 - <<'PY'
+import json, sys
+data = json.load(sys.stdin)
+print(str(data.get("present")))
+print(data.get("policy") or "")
+print(data.get("knownCount") if data.get("knownCount") is not None else 0)
+print(data.get("allowedCount") if data.get("allowedCount") is not None else 0)
+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
+    return 4
+  fi
+
+  # A herestring adds a newline, so mapfile would turn an empty parser result into one empty field.
+  # Keep that case separate so the refusal reports what the parser actually returned: zero fields.
+  if [ -z "$_FIELDS_RAW" ]; then
+    _FIELDS=()
+  else
+    mapfile -t _FIELDS <<< "$_FIELDS_RAW"
+  fi
+
+  # 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 fixed-field read in main() expects,
+  # whatever the reason: a producer that printed nothing, malformed JSON that jq/python3 still
+  # accepted, or a schema change upstream. main()'s slice (`_FIELDS[@]:5`) does not fire
+  # `set -u` on an unset OR a short array, and every fixed-field read there 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
+    return 5
+  fi
+}
+
+main() {
 set -uo pipefail
+# `pipefail` is not what catches the parser failure handled in parse_policy_fields() above (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"
@@ -121,31 +220,10 @@ EOF
   exit 1
 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.
-if command -v jq >/dev/null 2>&1; then
-  mapfile -t _FIELDS < <(printf '%s' "$POLICY_JSON" | jq -r '
-      (.present | tostring),
-      (.policy // ""),
-      (.knownCount // 0 | tostring),
-      (.allowedCount // 0 | tostring),
-      (.blockedCount // 0 | tostring),
-      (.known[]? // empty)')
-else
-  mapfile -t _FIELDS < <(printf '%s' "$POLICY_JSON" | python3 - <<'PY'
-import json, sys
-data = json.load(sys.stdin)
-print(str(data.get("present")))
-print(data.get("policy") or "")
-print(data.get("knownCount") if data.get("knownCount") is not None else 0)
-print(data.get("allowedCount") if data.get("allowedCount") is not None else 0)
-print(data.get("blockedCount") if data.get("blockedCount") is not None else 0)
-for n in (data.get("known") or []):
-    print(n)
-PY
-  )
-fi
+# Parse the policy in one pass and refuse on any of the three failure causes. The decision, the
+# three refusals and the reasoning behind each live in parse_policy_fields() above — kept there
+# with the code rather than here, so the explanation cannot drift away from what it explains.
+parse_policy_fields || exit $?
 
 PRESENT="${_FIELDS[0]:-null}"
 POLICY_MODE="${_FIELDS[1]:-}"
@@ -164,19 +242,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
@@ -251,3 +331,8 @@ How to read this:
     hardcoded list did. If the daemon's policy changes, the next run of this script reflects it
     with no edit to this file.
 EOF
+}
+
+if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
+  main "$@"
+fi

scripts/redeploy-fleetd.sh

@@ -178,6 +178,44 @@ swap_staged_jar() {
         may recover this once you find out why the move failed."
 }
 
+# fleetd #521 — the swap decision, and the step that acts on it.
+#
+# The defect: the swap step used to be guarded inline by `if [ "$DO_BUILD" = 1 ]` in the main flow.
+# Changing that to `if false` left the suite green and the swap never ran, so a redeploy reported
+# every step succeeding while the daemon started on no jar at all (stage_built_jar has already moved
+# the freshly built one to $JAR_STAGED by then) or on a stale one.
+# test_swap_ordered_after_wait_and_before_start could not catch it: it reads this script's own text
+# and compares line positions, and a same-line edit moves no line.
+#
+# Why these are TWO functions, and why the second one exists at all. Extracting only the predicate
+# — `should_swap`, which is what #521 asked for — is not enough, and this was measured, not guessed:
+# with the main flow calling `if should_swap "$DO_BUILD"; then`, changing THAT to `if false; then`
+# still left the whole suite at exit 0 with no failures. Tests that call a predicate directly prove
+# the predicate is right; nothing makes the code that does the work consult it. Extraction had moved
+# the untested decision one level up rather than removing it.
+#
+# So the decision and the action live together in swap_if_built, and the main flow has no guard of
+# its own to get wrong — it calls one function unconditionally. A test then calls swap_if_built with
+# both values of do_build and checks whether the swap actually happened, which fails if the guard is
+# removed, inverted, or stops being consulted. should_swap stays a separate predicate because it is
+# the decision itself and is worth naming and testing on its own.
+#
+# What this still does not pin: deleting the swap_if_built call from the main flow altogether. That
+# is the ordering test's job — its needle is that call site — and no test in this file can do better,
+# because sourcing stops before the main flow ever runs (see the SOURCED guard below).
+should_swap() {
+  local do_build="$1"
+  [ "$do_build" = 1 ]
+}
+
+swap_if_built() {
+  local do_build="$1"
+  should_swap "$do_build" || return 0
+  say "swap"
+  swap_staged_jar "$JAR_STAGED" "$JAR"
+  ok "jar in place: $(jar_id)"
+}
+
 # `launchctl list <label>` exits 0 iff the label is loaded (registered with launchd) — true whether
 # or not it is currently running, which is exactly "supervision is active" for our purposes. Read-
 # only: neither helper below changes anything, so both are also safe under --check.
@@ -452,6 +490,35 @@ classify_amqp_connection_errors() {
   REDEPLOY_UNEXPLAINED_ERRORS=$((REDEPLOY_UNEXPLAINED_ERRORS + pending_inbox + pending_lead_mailbox))
 }
 
+# fleetd #517: extracted so the suite can call this decision directly, the same way #510 extracted
+# wait_for_daemon_exit so its ordering became checkable. Before this, the only test of the drain-gate
+# abort message was a grep of this script's own source for the wording — so mutating the `if` below
+# to `if false` (making the branch unreachable) left every test green, because the wording was still
+# sitting in the file. Pure: only decides which message applies and prints it, no side effects, so a
+# test can call it directly with an in-memory staged path instead of driving the real drain-gate flow
+# (which needs a live $OLD_PID and an interactive prompt neither test can supply).
+#
+# The four cases:
+#   build ran,  staged jar present -> names the staged jar and how to finish or discard it
+#   build ran,  staged jar absent  -> "nothing changed" (nothing was staged this run either)
+#   --no-build, staged jar present -> ALSO "nothing changed", deliberately: --no-build itself builds
+#     and stages nothing (see require_no_build_jar above), so a staged jar found here is a leftover
+#     from an earlier, unrelated run. THIS run truly changed nothing, and the next DO_BUILD=1 run
+#     wipes that leftover before it builds (`rm -f "$JAR_STAGED"` in the build section above) — so
+#     there is nothing here for the operator to lose track of.
+#   --no-build, staged jar absent  -> "nothing changed"
+drain_gate_refusal() {
+  local do_build="$1" staged_path="$2"
+  if [ "$do_build" = 1 ] && [ -f "$staged_path" ]; then
+    printf 'aborted — the running daemon was NOT touched, but the freshly built jar is sitting at
+        %s, not yet swapped into %s. 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 %s by hand if you want to discard this build.' "$staged_path" "$JAR" "$staged_path"
+  else
+    printf 'aborted — nothing changed'
+  fi
+}
+
 # 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
@@ -610,15 +677,10 @@ if [ -n "$OLD_PID" ] && [ "$ASSUME_YES" = 0 ]; then
   echo
   read -r -p "   Fleet drained? type yes to restart: " reply
   if [ "$reply" != "yes" ]; then
-    # fleetd #493: "nothing changed" would be a lie once a build has run — the freshly built jar
-    # already moved to $JAR_STAGED (stage_built_jar, above), so the live path has one fewer file
-    # 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."
-    fi
-    die "aborted — nothing changed"
+    # fleetd #493 / #517: "nothing changed" would be a lie once a build has run and staged a jar —
+    # see drain_gate_refusal above for the full decision and why each of its four cases reads the
+    # way it does.
+    die "$(drain_gate_refusal "$DO_BUILD" "$JAR_STAGED")"
   fi
 fi
 
@@ -696,11 +758,7 @@ fi
 # NOW is it safe to put the freshly built jar at the path the NEXT `java -jar` (direct, or via
 # launchd/systemd's ExecStart) will read from — this mv is the one and only write to $JAR anywhere
 # in this script's mutating flow. If it fails, do not start: die() below exits before "start" runs.
-if [ "$DO_BUILD" = 1 ]; then
-  say "swap"
-  swap_staged_jar "$JAR_STAGED" "$JAR"
-  ok "jar in place: $(jar_id)"
-fi
+swap_if_built "$DO_BUILD"
 
 # ------------------------------------------------------------------ start
 # Unsupervised: login shell (zsh -l) is what puts the secrets on the daemon's environment, and cwd

scripts/test-probe-member-credentials.sh

@@ -0,0 +1,109 @@
+#!/usr/bin/env bash
+# Self-contained checks for the policy parsing guards in probe-member-credentials.sh.
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+PROBE="$ROOT/scripts/probe-member-credentials.sh"
+TMP="$(mktemp -d "$ROOT/.probe-member-credentials-test.XXXXXX")"
+trap 'rm -rf "$TMP"' EXIT
+
+# The SOURCED guard exposes this pure parser without contacting POLICY_URL.
+source "$PROBE"
+
+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"
+}
+
+assert_contains() {
+  local needle="$1" text="$2" description="$3"
+  printf '%s' "$text" | grep -qF "$needle" || fail "$description: missing $needle"
+}
+
+make_jq() {
+  local body="$1"
+  mkdir -p "$TMP/bin"
+  printf '%s\n' '#!/usr/bin/env bash' "$body" > "$TMP/bin/jq"
+  chmod +x "$TMP/bin/jq"
+}
+
+run_parser() {
+  local output rc=0
+  POLICY_JSON="$(< "$TMP/policy.json")"
+  POLICY_URL="fixture://member-credentials"
+  output="$(PATH="$TMP/bin:$PATH" parse_policy_fields 2>&1)" || rc=$?
+  PARSER_OUTPUT="$output"
+  PARSER_RC="$rc"
+}
+
+test_bash_older_than_four_refuses() {
+  local output rc=0 version
+  version="$(/bin/bash -c 'printf %s "$BASH_VERSION"')"
+  output="$(/bin/bash "$PROBE" 2>&1)" || rc=$?
+  assert_equals 3 "$rc" "bash 3 refusal status"
+  assert_contains 'This shell is bash' "$output" "bash 3 refusal"
+  assert_contains "$version" "$output" "bash 3 refusal version"
+}
+
+test_parser_non_zero_refuses() {
+  make_jq 'exit 17'
+  run_parser
+  assert_equals 4 "$PARSER_RC" "parser failure status"
+  assert_contains 'jq exited non-zero (status 17)' "$PARSER_OUTPUT" "parser failure message"
+}
+
+test_short_parser_output_refuses() {
+  make_jq "printf '%s\\n' true enforce 3 2"
+  run_parser
+  assert_equals 5 "$PARSER_RC" "short parser output status"
+  assert_contains 'policy parser (jq) returned 4 field(s)' "$PARSER_OUTPUT" "short parser output count"
+}
+
+test_empty_parser_output_reports_zero_fields() {
+  make_jq ':'
+  run_parser
+  assert_equals 5 "$PARSER_RC" "empty parser output status"
+  assert_contains 'policy parser (jq) returned 0 field(s)' "$PARSER_OUTPUT" "empty parser output count"
+}
+
+test_well_formed_policy_prints_name_table() {
+  local output rc=0
+  make_jq "cat '$TMP/policy.fields'"
+  # Shell functions cannot be passed in an environment assignment. Run the executable through bash.
+  output="$(BRIDGED_MEMBER=1 FIXTURE="$TMP/policy.json" PROBE="$PROBE" PATH="$TMP/bin:$PATH" bash -c '
+    curl() { cat "$FIXTURE"; }
+    export -f curl
+    exec "$PROBE"
+  ' 2>&1)" || rc=$?
+  assert_equals 0 "$rc" "well-formed policy status"
+  assert_contains 'ALPHA_TOKEN' "$output" "name table"
+  assert_contains 'BETA_TOKEN' "$output" "name table"
+  assert_contains 'GAMMA_TOKEN' "$output" "name table"
+}
+
+cat > "$TMP/policy.json" <<'JSON'
+{"present":true,"policy":"enforce","knownCount":3,"allowedCount":2,"blockedCount":1,"known":["ALPHA_TOKEN","BETA_TOKEN","GAMMA_TOKEN"]}
+JSON
+cat > "$TMP/policy.fields" <<'FIELDS'
+true
+enforce
+3
+2
+1
+ALPHA_TOKEN
+BETA_TOKEN
+GAMMA_TOKEN
+FIELDS
+
+test_bash_older_than_four_refuses
+test_parser_non_zero_refuses
+test_short_parser_output_refuses
+test_empty_parser_output_reports_zero_fields
+test_well_formed_policy_prints_name_table
+printf 'PASS: probe member credentials guards\n'

scripts/test-redeploy-fleetd.sh

@@ -206,6 +206,46 @@ 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 #517 — jar_id()'s "absent" branch was unpinned by any test: the existing test above (#511)
+# proves both halves of the present-file contract but never exercises the missing-file path. This
+# word matters more than a string usually would: "absent" is the #413 signal that a `mvn clean`
+# deleted the running daemon's jar out from under it, and the `redeploy-fleetd` skill points
+# operators at `--check` for exactly this. Covers both the no-argument default and an explicit path,
+# since the mutation (`absent` -> `present`) sits on the single shared `|| echo` and would flip both.
+test_jar_id_reports_absent_for_missing_file() {
+  local saved_jar="$JAR" dir default_result explicit_result
+  dir="$TMP/jar-id-absent"; mkdir -p "$dir"
+  JAR="$dir/does-not-exist.jar"
+  [ ! -f "$JAR" ] || fail "test fixture error: \$JAR unexpectedly exists at $JAR"
+  default_result="$(jar_id)"
+  explicit_result="$(jar_id "$dir/also-does-not-exist.jar")"
+  JAR="$saved_jar"
+  assert_equals "absent" "$default_result" "jar_id with no arguments must report absent when \$JAR does not exist"
+  assert_equals "absent" "$explicit_result" "jar_id with an explicit missing path must report absent"
+}
+
 # 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
@@ -325,23 +365,160 @@ test_wait_for_daemon_exit_times_out_if_pid_never_clears() {
   source "$ROOT/scripts/redeploy-fleetd.sh"   # restore the real running_pid/sleep for later tests
 }
 
+# fleetd #521 — the swap step's guard, at two levels.
+#
+# The first two tests call the predicate should_swap() directly. They pin its logic, and that is all
+# they pin. On their own they did NOT close #521, and this was measured rather than argued: with the
+# main flow reading `if should_swap "$DO_BUILD"; then`, changing that line to `if false; then` left
+# this whole suite at exit 0 with zero FAIL lines, because nothing here made the code that performs
+# the swap consult the predicate at all. Extracting the decision had moved the untested decision up
+# a level, not removed it.
+#
+# So the last two tests call swap_if_built() — the function the main flow actually calls, holding the
+# guard and the swap together — with a recording stub in place of the real `mv`. Those fail if the
+# guard is removed, inverted, or stops being consulted.
+#
+# What none of these four can catch: deleting the `swap_if_built "$DO_BUILD"` line from the main flow
+# altogether. That is test_swap_ordered_after_wait_and_before_start's job below, because sourcing
+# stops before the main flow runs, so no test in this file can invoke it.
+test_should_swap_true_when_build_ran() {
+  should_swap 1 || fail "should_swap 1 (a build ran and staged a jar) must return true"
+}
+
+test_should_swap_false_when_build_skipped() {
+  if should_swap 0; then
+    fail "should_swap 0 (--no-build; nothing was staged this run) must return false"
+  fi
+}
+
+# Both of these re-source redeploy-fleetd.sh at the START, because a bash function definition is
+# global for the rest of the process and an earlier test may have left swap_staged_jar or jar_id
+# overridden (see the longer note on this at test_detect_supervisor_systemd_probe_error_is_unclear),
+# and again at the END, so their own stubs do not leak into every test that runs after them.
+test_swap_if_built_performs_the_swap_when_build_ran() {
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+  local marker="$TMP/swap-if-built-ran"
+  rm -f "$marker"
+  swap_staged_jar() { printf '%s -> %s\n' "$1" "$2" > "$marker"; }
+  jar_id() { printf 'stubbed\n'; }
+  swap_if_built 1 > /dev/null
+  [ -f "$marker" ] \
+    || fail "swap_if_built 1 (a build ran and staged a jar) must perform the swap, and did not"
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+}
+
+test_swap_if_built_skips_the_swap_when_build_skipped() {
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+  local marker="$TMP/swap-if-built-skipped"
+  rm -f "$marker"
+  swap_staged_jar() { printf 'swapped\n' > "$marker"; }
+  jar_id() { printf 'stubbed\n'; }
+  swap_if_built 0 > /dev/null
+  if [ -f "$marker" ]; then
+    fail "swap_if_built 0 (--no-build; nothing was staged this run) must not swap, but it did"
+  fi
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+}
+
 # fleetd #493 item 2: "put the swap after that wait, before the start." Sourcing stops before the
 # main flow ever runs (see the SOURCED guard in redeploy-fleetd.sh), so the ordering guarantee
 # itself — as opposed to the pure functions it's built from — can only be checked by reading the
 # script's own call sites, the same way test_recovery_patterns_match_source below checks Java
 # source shape instead of behavior it cannot invoke directly.
+#
+# Two details about the three greps below, both of which have already gone wrong here.
+#
+# The needle for the swap is the MAIN FLOW's call site, `swap_if_built "$DO_BUILD"` — not
+# `swap_staged_jar "$JAR_STAGED" "$JAR"`. Since fleetd #521 that second string lives inside
+# swap_if_built's body, which is defined near the top of the script, far ABOVE the stop step. Using
+# it made this test report "swap_staged_jar (line 215) is not after wait_for_daemon_exit (line 730)"
+# — a true statement about a function definition, and nothing at all about the order of the steps.
+#
+# Each grep ends in `|| true`. This file runs under `set -euo pipefail`, and `pipefail` makes the
+# pipeline's status grep's status, so a needle that is simply ABSENT failed the assignment and `set
+# -e` killed the whole suite on the spot — before reaching the `[ -n ... ] || fail` line written to
+# report exactly that. Measured: the suite exited 1 having printed zero bytes, no FAIL line and no
+# name of the missing call site. `|| true` lets the assignment succeed empty so the guard can speak.
 test_swap_ordered_after_wait_and_before_start() {
   local src="$ROOT/scripts/redeploy-fleetd.sh" wait_line swap_line start_line
-  wait_line="$(grep -Fn 'wait_for_daemon_exit "$STOP_WAIT"' "$src" | head -1 | cut -d: -f1)"
-  swap_line="$(grep -Fn 'swap_staged_jar "$JAR_STAGED" "$JAR"' "$src" | head -1 | cut -d: -f1)"
-  start_line="$(grep -Fn 'say "start"' "$src" | head -1 | cut -d: -f1)"
+  wait_line="$(grep -Fn 'wait_for_daemon_exit "$STOP_WAIT"' "$src" | head -1 | cut -d: -f1 || true)"
+  swap_line="$(grep -Fn 'swap_if_built "$DO_BUILD"' "$src" | head -1 | cut -d: -f1 || true)"
+  start_line="$(grep -Fn 'say "start"' "$src" | head -1 | cut -d: -f1 || true)"
   [ -n "$wait_line" ] || fail "could not find the wait-for-exit call site in redeploy-fleetd.sh"
   [ -n "$swap_line" ] || fail "could not find the swap call site in redeploy-fleetd.sh"
   [ -n "$start_line" ] || fail "could not find the start section in redeploy-fleetd.sh"
   [ "$swap_line" -gt "$wait_line" ] \
-    || fail "swap_staged_jar (line $swap_line) is not after wait_for_daemon_exit (line $wait_line)"
+    || fail "swap_if_built (line $swap_line) is not after wait_for_daemon_exit (line $wait_line)"
   [ "$swap_line" -lt "$start_line" ] \
-    || fail "swap_staged_jar (line $swap_line) is not before the start section (line $start_line)"
+    || fail "swap_if_built (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"
+}
+
+# fleetd #517 — the drain-gate abort branch itself. Before this, the only test of this message was
+# a source-text grep (test_drain_gate_abort_message_says_no_no_build, below): it greps this script's
+# own file for the wording, which stays in the file even if the `if` guarding it is mutated to
+# `if false` and the branch can never run. These four tests call drain_gate_refusal directly instead,
+# so they fail if the branch is unreachable OR if its wording regresses — the grep test is KEPT
+# alongside these, not replaced, because it catches a different regression (a re-wording that still
+# reaches the right branch would not change which case fires here, but would still be worth pinning).
+test_drain_gate_refusal_build_ran_staged_present() {
+  local dir staged result
+  dir="$TMP/drain-refusal-build-staged"; mkdir -p "$dir"
+  staged="$dir/fleetd-new.jar"
+  printf 'staged jar bytes' > "$staged"
+  result="$(drain_gate_refusal 1 "$staged")"
+  printf '%s' "$result" | grep -qF "$staged" \
+    || fail "build-ran+staged-present refusal does not name the staged jar path"
+  printf '%s' "$result" | grep -qF 'Rerun WITHOUT --no-build' \
+    || fail "build-ran+staged-present refusal does not tell the operator how to finish the restart"
+  if printf '%s' "$result" | grep -qF 'nothing changed'; then
+    fail "build-ran+staged-present refusal must not claim nothing changed — the jar already moved"
+  fi
+}
+
+test_drain_gate_refusal_build_ran_staged_absent() {
+  local dir result
+  dir="$TMP/drain-refusal-build-no-staged"; mkdir -p "$dir"
+  result="$(drain_gate_refusal 1 "$dir/fleetd-new.jar")"
+  assert_equals "aborted — nothing changed" "$result" "build-ran+staged-absent refusal wording"
+}
+
+# --no-build itself never builds or stages anything (require_no_build_jar, above), so a staged jar
+# found here is a leftover from an earlier, unrelated run — THIS run truly changed nothing. See the
+# comment above drain_gate_refusal in redeploy-fleetd.sh for the full reasoning.
+test_drain_gate_refusal_no_build_staged_present() {
+  local dir staged result
+  dir="$TMP/drain-refusal-no-build-staged"; mkdir -p "$dir"
+  staged="$dir/fleetd-new.jar"
+  printf 'leftover staged jar bytes' > "$staged"
+  result="$(drain_gate_refusal 0 "$staged")"
+  assert_equals "aborted — nothing changed" "$result" "no-build+staged-present refusal must deliberately say nothing changed"
+}
+
+test_drain_gate_refusal_no_build_staged_absent() {
+  local dir result
+  dir="$TMP/drain-refusal-no-build-no-staged"; mkdir -p "$dir"
+  result="$(drain_gate_refusal 0 "$dir/fleetd-new.jar")"
+  assert_equals "aborted — nothing changed" "$result" "no-build+staged-absent refusal wording"
 }
 
 test_no_errors() {
@@ -555,16 +732,27 @@ 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_jar_id_reports_absent_for_missing_file
 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
 test_swap_staged_jar_dies_without_staged_file
 test_swap_staged_jar_dies_when_mv_fails
+test_should_swap_true_when_build_ran
+test_should_swap_false_when_build_skipped
+test_swap_if_built_performs_the_swap_when_build_ran
+test_swap_if_built_skips_the_swap_when_build_skipped
 test_require_no_build_jar_dies_when_absent
 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_drain_gate_refusal_build_ran_staged_present
+test_drain_gate_refusal_build_ran_staged_absent
+test_drain_gate_refusal_no_build_staged_present
+test_drain_gate_refusal_no_build_staged_absent
 test_no_errors
 test_recovery_patterns_match_source
 test_attributed_recovered_connection_error