#316: re-check worktree dirtiness after the pane stops, before removing it

SessionManager.releaseRemoved read hasUncommitted() once, while the worker
could still write, then used that stale boolean after launcher.stop() to
authorise `git worktree remove --force`. The same stale read also gated
trySnapshot, so a worker that wrote between the read and the stop lost its
work with neither a preserve nor a snapshot.

Add a second, best-effort hasUncommitted read immediately before the
removal, taken only on the path that is actually about to delete something
(never on a release that already decided to preserve, and never for
SHUTDOWN, which preserves unconditionally). If the tree is now dirty,
preserve it and attempt a fresh snapshot, since the original snapshot never
ran when the pre-stop read said clean. A failing re-check also preserves,
matching the existing CB-581 fail-safe rule.

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

@@ -22,6 +22,7 @@ import dev.ltms.fleet.placement.BackendQuarantine;
 import dev.ltms.fleet.placement.PlacementException;
 import dev.ltms.fleet.session.SessionManager;
 import dev.ltms.fleet.session.MemberSession;
+import dev.ltms.fleet.session.ShuttingDownException;
 import dev.ltms.fleet.session.WorktreeRequest;
 import dev.ltms.fleet.peer.MemberRole;
 import dev.ltms.fleet.peer.PeerLauncher;
@@ -962,6 +963,10 @@ public final class FleetMcp {
             return text(json(memberView(member)));
         } catch (GuardException e) {
             return error("subscription boundary: " + e.getMessage());
+        } catch (ShuttingDownException e) {
+            // fleetd #308: the daemon's shutdown drain has already started — refuse loudly rather
+            // than register a session drainAll will never see again.
+            return error("shutting down: " + e.getMessage());
         } catch (PlacementException e) {
             // CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — distinct
             // from "profile does not exist" below.

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

@@ -385,9 +385,12 @@ public final class CompositePeerLauncher implements PeerLauncher {
             }
         }
 
+        // unreachable.size() counts DISTINCT profiles, not attempts (a HashSet dedupes a profile
+        // added twice) — say "distinct" so the count matches the sentence and the profile list that
+        // follows, rather than reading as a count of attempts made (fleetd #315).
         throw new PeerUnreachableException(
                 "no reachable worker profile available after trying " + unreachable.size()
-                        + " candidate(s): " + String.join(", ", unreachable));
+                        + " distinct candidate(s): " + String.join(", ", unreachable));
     }
 
     /**

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

@@ -187,6 +187,24 @@ public final class MessageService {
         private volatile Long completedNanos;
         private volatile Reply question;
         private volatile String turnId;
+        /**
+         * Set when this task's {@code fleet_ask} lapsed with no answer (fleetd #307):
+         * {@link #clearAsyncQuestion} then forgets {@link #turnId} (nulls it and drops the task from
+         * {@code asyncTasksByTurn}) so {@link #hasAsyncQuestion} stops reporting the target BUSY — a
+         * later {@code fleet_send} to it must be accepted, not refused. But the worker's turn is
+         * still genuinely live: it resumed on its own and will eventually call its real
+         * {@code fleet_reply}. Losing {@link #turnId} loses {@link #askAnsweredAsyncTasks}' only
+         * signal that such a reply belongs to this task, so that reply used to fall straight to the
+         * inbox and strand — {@code fleet_poll} stayed {@code PENDING} forever, later force-failed by
+         * {@link #abandon} with the misleading "session released before it replied". This flag is a
+         * second, independent signal that survives the forgetting: {@link #askAnsweredAsyncTasks}
+         * accepts it in place of a live {@link #turnId}, without ever re-adding the task to
+         * {@code asyncTasksByTurn} (so the BUSY release is untouched). Cleared implicitly once
+         * {@link #future} resolves — every match in {@link #askAnsweredAsyncTasks} already requires
+         * {@code !future.isDone()}, so a task that recovered (or was later failed by
+         * {@link #abandon}) can never match again regardless of this flag's value.
+         */
+        private volatile boolean askTimedOut;
 
         private Task(String ticket, String target, LongSupplier nowNanos) {
             this.ticket = ticket;
@@ -395,18 +413,16 @@ public final class MessageService {
      * {@link Rendezvous#resolveQuestion} must keep today's {@code NO_WAITER} behaviour — questions
      * are interactive and must never be queued.
      *
-     * <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks}
-     * cannot actually return more than one entry today (see its own javadoc for why — in short,
-     * {@link #hasAsyncQuestion} keeps a target BUSY, so no second task can reach this state, for as
-     * long as an earlier one's {@code turnId} is still stamped). That is an emergent guarantee from
-     * two other facts, not one this method enforces, so this branch stays in as defence in depth
-     * rather than being removed as dead code: if it ever weakens, returning whichever candidate a
-     * {@code ConcurrentHashMap} iteration reaches first would let a genuine reply complete the
-     * <em>wrong</em> ticket — silently handing the lead something that reads like a correct answer to
-     * a delegation the worker never touched, which is worse than a failure because the lead acts on
-     * it. When more than one candidate exists, guessing is not safe: fall back to the inbox exactly
-     * as the zero-candidate case does, and let {@link #abandon} apply the eventual recovery
-     * deterministically instead.
+     * <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks} can
+     * return more than one entry — a reachable state, not a hypothetical one (see its own javadoc:
+     * an {@code fleet_ask} that lapsed with no answer, fleetd #307, frees the target for a completely fresh
+     * delegation, which can itself go on to ask-and-lapse before the first worker's real reply
+     * arrives). Returning whichever candidate a {@code ConcurrentHashMap} iteration reaches first
+     * would let a genuine reply complete the <em>wrong</em> ticket — silently handing the lead
+     * something that reads like a correct answer to a delegation the worker never touched, which is
+     * worse than a failure because the lead acts on it. When more than one candidate exists, guessing
+     * is not safe: fall back to the inbox exactly as the zero-candidate case does, and let
+     * {@link #abandon} apply the eventual recovery deterministically instead.
      *
      * <p><strong>{@code content} is required (fleetd #302).</strong> Both doors that reach this
      * method must reject a missing/blank reply the same way, so the check lives here rather than in
@@ -434,15 +450,18 @@ public final class MessageService {
             count(FleetMetrics.REPLIES, "path", "rendezvous");
             return true;                       // a live send took it — unchanged fast path
         }
-        // #137: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming a
-        // turn that {@link #answer} already gave up waiting on. answer()'s own bounded wait (the
-        // primary's fleet_send{turnId} call, capped well under a minute) can time out and close its
-        // waiter long before the worker — now actually resuming real work — finishes and replies. That
-        // reply used to have nowhere to land but the session inbox, leaving the async ticket's future
-        // unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's abandon() forced it
-        // FAILED with a misleading "session released before it replied" reason, even though the reply
-        // had, in fact, arrived. Completing the matching ticket directly here means fleet_poll{ticket}
-        // sees the real reply instead.
+        // #137/fleetd #307: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming
+        // a turn that either answer() (#137) or ask() (fleetd #307) already gave up waiting on:
+        //  - answer()'s own bounded wait (the primary's fleet_send{turnId} call, capped well under a
+        //    minute) can time out and close its waiter long before the worker — now actually resuming
+        //    real work — finishes and replies.
+        //  - ask()'s own wait for the primary can time out first, with the worker resuming on its own
+        //    and finishing unanswered.
+        // Either way that reply used to have nowhere to land but the session inbox, leaving the async
+        // ticket's future unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's
+        // abandon() forced it FAILED with a misleading "session released before it replied" reason,
+        // even though the reply had, in fact, arrived. Completing the matching ticket directly here
+        // means fleet_poll{ticket} sees the real reply instead.
         List<Task> candidates = askAnsweredAsyncTasks(session);
         if (candidates.size() == 1) {
             Task orphan = candidates.get(0);
@@ -473,34 +492,42 @@ public final class MessageService {
     }
 
     /**
-     * Every still-open async task on {@code target} whose {@code fleet_ask} was already answered —
-     * its {@link Task#turnId} is stamped but its {@link Task#question} was cleared by {@link #answer}
-     * — yet whose future is not resolved yet (#137). Empty if no such task exists, including the
-     * common case where {@code target}'s worker never used {@code fleet_ask} at all (a task that was
-     * never asked has {@code turnId == null}, so it can never match here and only ever completes
-     * through the ordinary rendezvous fast path in {@link #reply}).
+     * Every still-open async task on {@code target} whose worker is genuinely expected to send a
+     * real {@code fleet_reply} next with nothing left registered to catch it: either its
+     * {@code fleet_ask} was already answered — {@link Task#turnId} is stamped but {@link
+     * Task#question} was cleared by {@link #answer} — or its {@code fleet_ask} lapsed unanswered and
+     * {@link Task#askTimedOut} marks that (fleetd #307; {@link Task#turnId} is {@code null} by then, forgotten
+     * so the target is not left BUSY — see {@link Task#askTimedOut}'s own javadoc). Either way the
+     * task's future is not resolved yet. Empty if no such task exists, including the common case
+     * where {@code target}'s worker never used {@code fleet_ask} at all (a task that was never asked
+     * has both {@code turnId == null} and {@code askTimedOut == false}, so it can never match here and
+     * only ever completes through the ordinary rendezvous fast path in {@link #reply}).
      *
-     * <p><strong>Returns at most one entry today — verified, not assumed.</strong> {@link #send}
-     * refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is true, and that
-     * check matches ANY task whose {@code turnId} is still stamped in {@code asyncTasksByTurn} —
-     * not only while its question is still open. {@link #answer} deliberately leaves that stamp in
-     * place ({@code clearAsyncQuestion(turnId, false)}) until the resumed turn's own future actually
-     * resolves, at which point {@link #finishAsyncTask} both removes the stamp AND completes that
-     * task's future in the same call. So a second task can never reach "{@code turnId} stamped, future
-     * still open" — the exact pair this method matches on — while a first one already holds it: by
-     * the time the stamp is gone, so is the eligibility. This is an emergent property of those two
-     * facts holding together, not something this method (or its callers) enforces on its own — flip
-     * {@code forgetTurn} to {@code true} in that one {@link #answer} call and it silently stops being
-     * true, with nothing left to fail loudly. The callers below still handle "more than one" as
-     * defence in depth against exactly that, not because they exercise it today: {@link #reply}
-     * treats it as unresolvable and falls back to the inbox; {@link #abandon} would pick the oldest
-     * deterministically (its own {@code matching} list has no such guarantee — see its javadoc).
+     * <p><strong>Can return more than one entry — reachable, not just defence in depth.</strong>
+     * {@link #send} refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is
+     * true, and that check matches ANY task whose {@code turnId} is still stamped in
+     * {@code asyncTasksByTurn}. While a task's {@code turnId} stays stamped — {@link #answer} leaves
+     * it in place ({@code clearAsyncQuestion(turnId, false)}) until {@link #finishAsyncTask} removes
+     * the stamp and completes the future in the same call — no second task on the same target can
+     * reach an eligible state, because {@link #send} would refuse it as BUSY first. That single-task
+     * guarantee holds only for the {@code turnId}-stamped half of this method's match: an
+     * {@link Task#askTimedOut} task is, by construction, no longer stamped in {@code asyncTasksByTurn}
+     * (that is the whole point of forgetting {@code turnId} in {@link #clearAsyncQuestion}), so the
+     * target is free the moment one ask lapses. A fresh, independent {@code sendAsync} to the same
+     * target can then be dispatched, itself pause on {@code fleet_ask}, and itself time out — landing
+     * a second {@code askTimedOut} task on the very target the first one is still waiting to answer
+     * for. Two (or more) genuinely open tasks on one target is therefore a real, reachable state
+     * today, not a hypothetical: {@link #reply} treats it as unresolvable and falls back to the
+     * inbox rather than guess which task a reply belongs to (guessing wrong would hand the lead a
+     * plausible-looking answer to a delegation the worker never touched — worse than a failure,
+     * because the lead acts on it); {@link #abandon} instead picks the oldest deterministically (its
+     * own {@code matching} list has a different, wider match — see its javadoc).
      */
     private List<Task> askAnsweredAsyncTasks(String target) {
         List<Task> candidates = new ArrayList<>();
         for (Task task : tasks.values()) {
-            if (target.equals(task.target) && task.question == null && task.turnId != null
-                    && !task.future.isDone()) {
+            if (target.equals(task.target) && task.question == null && !task.future.isDone()
+                    && (task.turnId != null || task.askTimedOut)) {
                 candidates.add(task);
             }
         }
@@ -898,6 +925,12 @@ public final class MessageService {
             return new AskResult(AskOutcome.ANSWERED, answer);
         } catch (TimeoutException e) {
             log.debug("fleet_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
+            // fleetd #307: mark the task BEFORE clearAsyncQuestion(forgetTurn=true) below drops it out of
+            // asyncTasksByTurn and nulls its turnId — that forgetting is deliberate and stays (it is
+            // what keeps the target from staying BUSY forever), but it would otherwise also erase
+            // askAnsweredAsyncTasks' only signal that the worker's eventual real fleet_reply still
+            // belongs to this task, stranding it in the inbox with a false "never replied" verdict.
+            markAskTimedOut(ticket.turnId());
             clearAsyncQuestion(ticket.turnId(), true);
             return new AskResult(AskOutcome.TIMED_OUT, null);
         } catch (ExecutionException e) {
@@ -1162,6 +1195,23 @@ public final class MessageService {
         return task;
     }
 
+    /**
+     * Mark {@code turnId}'s task as having a {@code fleet_ask} that lapsed with no answer (fleetd #307), so
+     * {@link #askAnsweredAsyncTasks} still recognizes the worker's eventual real {@code fleet_reply}
+     * as belonging to it after {@link #clearAsyncQuestion}'s {@code forgetTurn=true} erases
+     * {@link Task#turnId} — see {@link Task#askTimedOut}. Must be called before that forgetting, while
+     * {@code turnId} can still resolve the task in {@code asyncTasksByTurn}; a lookup afterward would
+     * find nothing. Only when it matches the task's current turn — same guard as
+     * {@link #clearAsyncQuestion} — so a chained second {@code fleet_ask} (#282) that already moved
+     * the task to a fresh {@code turnId} cannot mark it for a turn that is no longer its own.
+     */
+    private void markAskTimedOut(String turnId) {
+        Task task = asyncTasksByTurn.get(turnId);
+        if (task != null && turnId.equals(task.turnId)) {
+            task.askTimedOut = true;
+        }
+    }
+
     /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
     private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
         // CB-582: tell the push loop first — like ticketCollected, a removal for a turnId it never

fleetd/src/main/java/dev/ltms/fleet/placement/FixedPlacementPolicy.java

@@ -5,10 +5,14 @@ import java.util.List;
 
 /**
  * Backward-compatible placement: an unqualified spawn always resolves to the configured default
- * profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps and
- * reachability so that a pre-existing config behaves identically after upgrade.
+ * profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps
+ * ({@code maxLoad}) so that a pre-existing config behaves identically after upgrade — capacity
+ * gating for automatic placement is deliberately out of scope for {@code fixed}, exactly as it
+ * always has been. Reachability is a narrower exception (fleetd #315, below): a profile is never
+ * checked for reachability up front, only skipped once it has already failed in <em>this same</em>
+ * spawn call's retry loop — see the unreachable case below.
  *
- * <p>Three exceptions walk past the default instead of returning it unconditionally:
+ * <p>Four exceptions walk past the default instead of returning it unconditionally:
  * <ul>
  *   <li>Quarantine (CB-578 stage B): a quarantined default is a credential that just refused on
  *       a usage limit, not a transient capacity or reachability concern.
@@ -16,13 +20,21 @@ import java.util.List;
  *       ({@code BackendOutagePolicy}) — a separate, shorter-lived source from quarantine. When a
  *       profile is both quarantined and cooling off, only the quarantine reason is reported
  *       (exhaustion takes priority), matching {@code CompositePeerLauncher}'s explicit-spawn order.
+ *   <li>Unreachable (fleetd #315): {@code CompositePeerLauncher.spawn} retries a failed candidate
+ *       on the next one and rebuilds the {@link PlacementContext} so {@code ctx.unreachable()}
+ *       names every profile that already failed with {@code PeerUnreachableException} in this same
+ *       call. Without this check {@code select} kept handing back the same dead default forever —
+ *       the retry loop's own comment says "so the policy excludes this profile", and this is what
+ *       makes that true for {@code fixed} too, matching {@code weighted}/{@code round-robin}
+ *       (both filter on {@code ctx.unreachable()} via {@link PlacementPolicyUtil#available}).
  *   <li>Weight 0 (CB-554): {@code fixed} is still automatic selection, so a profile the operator
  *       marked "never auto-select me" ({@code weight <= 0}) must be skipped here exactly as
  *       {@code weighted}/{@code round-robin} skip it — an explicit {@code fleet_spawn} naming
  *       the profile is unaffected, only this automatic fallback walk.
  * </ul>
- * A fleet where nothing is ever quarantined, cooling off, or weight-0 never exercises any of these
- * paths, so today's behaviour is unchanged.
+ * A fleet where nothing is ever quarantined, cooling off, unreachable, or weight-0 never exercises
+ * any of these paths, so today's behaviour is unchanged — in particular, the very first selection
+ * of a spawn call always sees an empty {@code unreachable} set, so the first choice is untouched.
  */
 final class FixedPlacementPolicy implements PlacementPolicy {
 
@@ -30,12 +42,12 @@ final class FixedPlacementPolicy implements PlacementPolicy {
     public PlacementCandidate select(PlacementContext ctx) {
         String d = ctx.defaultProfile();
         if (d != null && !d.isBlank() && !ctx.quarantined().contains(d) && !ctx.coolingOff().contains(d)
-                && !weightExcluded(ctx, d)) {
+                && !ctx.unreachable().contains(d) && !weightExcluded(ctx, d)) {
             return new PlacementCandidate(d, null, 1.0f, null);
         }
         for (PlacementCandidate c : ctx.candidates()) {
             if (!ctx.quarantined().contains(c.profile()) && !ctx.coolingOff().contains(c.profile())
-                    && !c.excluded()) {
+                    && !ctx.unreachable().contains(c.profile()) && !c.excluded()) {
                 return new PlacementCandidate(c.profile(), null, c.weight(), c.maxLoad());
             }
         }
@@ -44,8 +56,9 @@ final class FixedPlacementPolicy implements PlacementPolicy {
             // Exhaustion quarantine takes priority: reported only when quarantine is absent, so the
             // message never claims "cooling off" for a profile that is really backend-exhausted.
             boolean dCoolingOff = !dQuarantined && ctx.coolingOff().contains(d);
+            boolean dUnreachable = ctx.unreachable().contains(d);
             boolean dWeightExcluded = weightExcluded(ctx, d);
-            if (dQuarantined || dCoolingOff || dWeightExcluded) {
+            if (dQuarantined || dCoolingOff || dUnreachable || dWeightExcluded) {
                 List<String> reasons = new ArrayList<>();
                 if (dQuarantined) {
                     reasons.add("is quarantined (backend exhausted)");
@@ -53,6 +66,9 @@ final class FixedPlacementPolicy implements PlacementPolicy {
                 if (dCoolingOff) {
                     reasons.add("is cooling off after repeated backend errors");
                 }
+                if (dUnreachable) {
+                    reasons.add("is unreachable");
+                }
                 if (dWeightExcluded) {
                     reasons.add("has weight 0 (excluded from automatic selection)");
                 }
@@ -62,7 +78,7 @@ final class FixedPlacementPolicy implements PlacementPolicy {
         }
         if (!ctx.candidates().isEmpty()) {
             throw new PlacementException("all worker profiles are excluded from automatic "
-                    + "selection (quarantined, cooling off, or weight-0)");
+                    + "selection (quarantined, cooling off, unreachable, or weight-0)");
         }
         throw new PlacementException("no worker profiles configured");
     }

fleetd/src/main/java/dev/ltms/fleet/rest/FleetApp.java

@@ -19,6 +19,7 @@ import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.placement.PlacementException;
 import dev.ltms.fleet.msg.MessageService;
 import dev.ltms.fleet.session.SessionManager;
+import dev.ltms.fleet.session.ShuttingDownException;
 import dev.ltms.fleet.peer.MemberRole;
 import dev.ltms.fleet.session.MemberSession;
 import dev.ltms.fleet.session.WorktreeRequest;
@@ -501,6 +502,11 @@ public final class FleetApp {
             ctx.status(201).json(view(member));
         } catch (GuardException e) {
             ctx.status(403).json(Map.of("error", "subscription_boundary", "detail", e.getMessage()));
+        } catch (ShuttingDownException e) {
+            // fleetd #308: the daemon's shutdown drain has already started — 503, not a bare 500,
+            // so this reads the same as PlacementException below: valid request, refused because
+            // of a transient daemon state rather than a bad argument.
+            ctx.status(503).json(Map.of("error", "shutting_down", "detail", e.getMessage()));
         } catch (PlacementException e) {
             // CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — a benign,
             // likely-transient refusal, distinct from "profile does not exist" below. 503: the

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

@@ -21,6 +21,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
@@ -61,6 +62,8 @@ public final class SessionManager implements TurnListener {
     private final LongSupplier nowNanos;
     private final int contextCap;
     private final boolean clearAfterTurn;
+    /** Null in production; test seam for the interval before an idle session's conditional release. */
+    private final Consumer<MemberSession> beforeIdleRelease;
     private volatile MemberLifecycle memberLifecycle = MemberLifecycle.NONE;
     /**
      * CB-586: the repo root the fleet actually works in, remembered the first time a worktree
@@ -71,6 +74,16 @@ public final class SessionManager implements TurnListener {
      */
     private volatile String fleetRepoRoot;
 
+    /**
+     * fleetd #308: flips true the instant {@link #drainAll} starts, before its registry snapshot
+     * is even taken — so a spawn already in flight sees the refusal as early as a plain flag can
+     * make it. This alone cannot close the race completely: a caller that read {@code false} just
+     * before the flip can still land in the registry after the snapshot. {@link #drainAll}'s
+     * post-loop sweep is what catches that straggler; the two mechanisms are deliberately paired,
+     * see {@link #drainAll}'s javadoc.
+     */
+    private final AtomicBoolean draining = new AtomicBoolean(false);
+
     /** CB-520: notified with a terminalId on every acquire; no-op until wired. */
     private final List<Consumer<String>> acquireListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
     /** CB-516: notified with a {@link ReleaseDetail} on every release; no-op until wired. */
@@ -102,13 +115,23 @@ public final class SessionManager implements TurnListener {
     }
 
     public SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
-                          int contextCap, boolean clearAfterTurn) {
+                           int contextCap, boolean clearAfterTurn) {
+        this(launcher, worktrees, nowNanos, contextCap, clearAfterTurn, null);
+    }
+
+    /**
+     * Package-private constructor for a deterministic reap/delivery race test. Production callers
+     * use the constructor above, whose null hook adds no callback or lock to an ordinary reap.
+     */
+    SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
+                   int contextCap, boolean clearAfterTurn, Consumer<MemberSession> beforeIdleRelease) {
         this.launcher = launcher;
         this.worktrees = worktrees;
         this.presence = new PresenceFleet(this);
         this.nowNanos = nowNanos;
         this.contextCap = contextCap;
         this.clearAfterTurn = clearAfterTurn;
+        this.beforeIdleRelease = beforeIdleRelease;
     }
 
     /**
@@ -181,6 +204,14 @@ public final class SessionManager implements TurnListener {
     public MemberSession acquire(String profile, MemberRole role, String requestedCwd, String callerCwd,
                                  String ownerTerminal, WorktreeRequest wt,
                                  String sessionName, String resumeSessionId) {
+        // fleetd #308: refuse before anything else runs — no slot reservation, no launcher spawn —
+        // so a caller learns the daemon is going down instead of getting a session drainAll will
+        // never see again. Checked here because every other acquire(...) overload delegates to
+        // this one, so this is the single point every spawn path passes through.
+        if (draining.get()) {
+            throw new ShuttingDownException("fleetd is shutting down; refusing to spawn a session "
+                    + "the shutdown drain would never see");
+        }
         MemberRole memberRole = (role == null) ? MemberRole.DEV : role;
         requireResumeCapability(profile, resumeSessionId);
         // CB-619 / fleetd #123: an explicit profile bypasses placement (CompositePeerLauncher only
@@ -272,10 +303,32 @@ public final class SessionManager implements TurnListener {
      */
     private void release(String paneId, ReleaseCause cause) {
         MemberSession removed = registry.remove(paneId);
+        releaseRemoved(paneId, removed, handles.remove(paneId), cause);
+    }
+
+    /**
+     * Tear a session down only while {@code expected} is still its registry value. A lifecycle
+     * transition replaces the immutable record, so this prevents a reap based on an old READY or
+     * DONE record from stopping a worker that delivery has made BUSY.
+     */
+    private boolean releaseIfCurrent(MemberSession expected, ReleaseCause cause) {
+        if (!registry.remove(expected.paneId(), expected)) {
+            // A lifecycle transition replaced the record between the caller's check and this remove.
+            // Log it: this race is by definition unobservable otherwise, and a reaper that silently
+            // declines to reap is the hardest kind of behaviour to diagnose after the fact.
+            log.debug("skipping reap of pane={}: its registry record changed after the idle check "
+                    + "(most likely a delivery made it BUSY)", expected.paneId());
+            return false;
+        }
+        releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
+        return true;
+    }
+
+    private void releaseRemoved(String paneId, MemberSession removed, PeerHandle removedHandle,
+                                ReleaseCause cause) {
         // fleetd #209: remove right alongside the registry entry so a released session's handle is
         // never leaked — but keep the local reference below, so the id can still be resolved for
         // the ReleaseDetail this teardown notifies with.
-        PeerHandle removedHandle = handles.remove(paneId);
         boolean preserveWorktree = cause == ReleaseCause.SHUTDOWN;
         String snapshotRef = null;
         if (removed != null) {
@@ -333,6 +386,31 @@ public final class SessionManager implements TurnListener {
         // from the registry with no pane stop is an orphaned pane — a live terminal burning a fleet
         // slot that no longer appears in the roster and can never be reclaimed.
         launcher.stop(paneId);
+        if (removed != null && !preserveWorktree && removed.worktree() != null) {
+            // fleetd #316: the `dirty` read above ran while the worker could still write to this
+            // worktree, so a stale `false` must not be trusted to authorise the --force removal
+            // below. Re-read the worktree's state one more time, right here — immediately before
+            // the one step that would destroy it, and only on the path that is actually about to
+            // do that (invariant 4: no second unconditional `git status` on a release that already
+            // decided to preserve). By now `launcher.stop` has returned, so this read reflects
+            // whatever the worker managed to write up to and including its teardown, not whatever
+            // it had written at release-start time.
+            if (dirtyImmediatelyBeforeRemoval(removed)) {
+                preserveWorktree = true;
+                // The pre-stop snapshot above never ran for this session (the pre-stop read said
+                // clean), so this is the only chance to get the newly-discovered work into
+                // refs/wip/* rather than leaving the on-disk preserve as the sole copy. Best-effort,
+                // like every other snapshot attempt — trySnapshot logs and swallows its own failure.
+                String lateSnapshotRef = trySnapshot(removed, cause);
+                log.warn("release {} preserves worktree {} for pane={} terminal={}: it reported "
+                        + "clean before the pane stopped but dirty immediately before removal — the "
+                        + "worker wrote to it during teardown, and --force removing it now would "
+                        + "have destroyed that work{}",
+                        cause, removed.worktree(), paneId, removed.terminalId(),
+                        lateSnapshotRef == null ? "" : " (snapshotted to refs/wip/" + removed.branch()
+                                + " commit=" + lateSnapshotRef + ")");
+            }
+        }
         if (removed != null && !preserveWorktree && removed.worktree() != null) {
             // fleetd #283: this is the one cleanup step in this method that used to be bare. By the
             // time it runs, the registry entry, the retained handle, and the pane are all already
@@ -351,6 +429,24 @@ public final class SessionManager implements TurnListener {
         }
     }
 
+    /**
+     * fleetd #316: the read that actually authorises {@code worktrees.remove}, taken with the
+     * worker's pane already stopped. Fails toward preserving (returns {@code true}) on any
+     * exception — the same rule the pre-stop check applies (CB-581): once we can no longer tell
+     * whether the worktree is dirty, preserving costs disk while deleting on a guess can destroy
+     * work that has no other copy.
+     */
+    private boolean dirtyImmediatelyBeforeRemoval(MemberSession removed) {
+        try {
+            return worktrees.hasUncommitted(removed.worktree());
+        } catch (RuntimeException e) {
+            log.warn("release could not re-check worktree {} for pane={} terminal={} immediately "
+                    + "before removal; preserving it rather than risk destroying unsaved work: {}",
+                    removed.worktree(), removed.paneId(), removed.terminalId(), e.toString());
+            return true;
+        }
+    }
+
     /**
      * Best-effort snapshot of a dirty worktree into {@code refs/wip/<branch>} (CB-578 stage C). A
      * failure here must never escalate: the caller has already decided to preserve the worktree
@@ -846,14 +942,18 @@ public final class SessionManager implements TurnListener {
             }
             long idleNanos = now - s.lastActivityAtNanos();
             if (idleNanos > idleTtlNanos) {
-                log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
-                        s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
-                        TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
                 // CB-581: one session that fails to release must not abort the whole reaping pass —
                 // match drainAll's per-session try/catch so the rest of the roster still gets reaped.
                 try {
-                    release(s.paneId());
-                    reaped++;
+                    if (beforeIdleRelease != null) {
+                        beforeIdleRelease.accept(s);
+                    }
+                    if (releaseIfCurrent(s, ReleaseCause.COMPLETED)) {
+                        log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
+                                s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
+                                TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
+                        reaped++;
+                    }
                 } catch (RuntimeException e) {
                     log.warn("reap failed for pane={} terminal={} worktree={}; continuing with "
                             + "remaining sessions", s.paneId(), s.terminalId(), s.worktree(), e);
@@ -884,10 +984,40 @@ public final class SessionManager implements TurnListener {
      * a reason to delete a worker's only copy of its uncommitted work. A session still {@code BUSY}
      * when the timeout expired is abandoned mid-turn and logged loudly so an operator can find its
      * kept worktree.
+     *
+     * <p>fleetd #308: {@code roster()} is a one-shot snapshot (see its javadoc), and nothing used
+     * to stop a new session from registering after it was taken — {@link #acquire} stayed open for
+     * as long as this drain waited on a {@code BUSY} session, up to the whole {@code timeoutNanos}
+     * budget. Two things close that window, deliberately paired because neither alone is complete:
+     * {@link #draining} is flipped true before the snapshot is even taken, so {@link #acquire}
+     * refuses (invariant 3: loudly, via {@link ShuttingDownException}) as much of the window as a
+     * plain flag can close; and the sweep below re-reads the registry once the initial snapshot has
+     * fully drained and drains whatever a straggler — a caller that read the flag as {@code false}
+     * a moment before it flipped — still managed to register. The sweep shares the same
+     * {@code deadline} rather than getting its own: {@code timeoutNanos} is a budget for the WHOLE
+     * 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.
      */
     void drainAll(long timeoutNanos) {
         long deadline = System.nanoTime() + timeoutNanos;
-        for (MemberSession s : roster()) {
+        draining.set(true);
+        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);
+        }
+    }
+
+    /**
+     * 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.
+     */
+    private void drainSnapshot(List<MemberSession> snapshot, long deadline) {
+        for (MemberSession s : snapshot) {
             try {
                 if (s.state() == MemberSession.State.BUSY) {
                     while (System.nanoTime() < deadline) {

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

@@ -0,0 +1,18 @@
+package dev.ltms.fleet.session;
+
+/**
+ * Thrown by {@link SessionManager#acquire} when a spawn is requested after the daemon's shutdown
+ * drain has already begun (fleetd #308).
+ *
+ * <p>{@link SessionManager#drainAll} snapshots the registry once and tears down exactly what is
+ * in that snapshot. A session registered after the snapshot is invisible to the drain loop: its
+ * pane is left running and its worktree is never preserved, and nothing else ever reclaims
+ * either — the daemon's in-memory registry dies with the process. Refusing the spawn here,
+ * loudly, is what stops that session from ever being created in the first place, rather than
+ * silently handing the caller a session the daemon can no longer manage.
+ */
+public final class ShuttingDownException extends RuntimeException {
+    public ShuttingDownException(String message) {
+        super(message);
+    }
+}

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

@@ -189,7 +189,7 @@ class FleetMcpTest {
     }
 
     @Test
-    void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
+    void unansweredAsyncAskReturnsTheTicketToPendingThenAWorkersLateReplyStillCompletesIt() throws Exception {
         McpSchema.CallToolResult accepted = FleetMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
         String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
 
@@ -203,8 +203,15 @@ class FleetMcpTest {
         assertTrue(textOf(ask).contains("no answer"), textOf(ask));
         assertTrue(textOf(FleetMcp.poll(messages, ticket, null)).startsWith("[pending"));
 
+        // fleetd #307: the worker resumed on its own after the primary never answered, and its real
+        // fleet_reply must complete its OWN async ticket — not strand in the inbox with
+        // fleet_poll{ticket} stuck PENDING forever and later force-failed with a false "session
+        // released before it replied" reason. This used to land in the inbox instead (see the old
+        // assertion this replaced: messages.drainReplies("term_a").getFirst()...) — that was the bug.
         FleetMcp.reply(messages, "term_a", "finished after timeout");
-        assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
+        assertEquals("finished after timeout", textOf(FleetMcp.poll(messages, ticket, null)));
+        assertTrue(messages.drainReplies("term_a").isEmpty(),
+                "the reply completed its own ticket directly and never touched the inbox");
     }
 
     @Test

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

@@ -615,6 +615,67 @@ class CompositePeerLauncherTest {
         assertEquals(1, adapter.spawnCount("b"));
     }
 
+    /**
+     * fleetd #315: {@code CompositePeerLauncher.spawn} rebuilds the {@link PlacementContext} after
+     * every failed attempt "so the policy excludes this profile" (see the comment at the retry call
+     * site) — but {@code FixedPlacementPolicy} never read {@code ctx.unreachable()}, so under the
+     * default {@code fixed} placement every retry re-picked the same dead default and a second,
+     * healthy, configured profile was never tried. This is the same scenario as
+     * {@link #failoverRetriesNextCandidateWhenProfileIsUnreachable}, but pinned to {@code fixed()}
+     * instead of {@code weighted()} — the three existing failover tests all use {@code weighted()},
+     * which is exactly why nobody caught this: the retry loop's contract has no coverage under its
+     * own default policy.
+     */
+    @Test
+    void failoverRetriesNextCandidateUnderFixedPlacementWhenProfileIsUnreachable() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, FleetConfig.Profile> profiles = ordered(
+                "a", stubWorker("a"),
+                "b", stubWorker("b"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "a", Set.of("a"));
+        CompositePeerLauncher composite = new CompositePeerLauncher(
+                List.of(adapter), "a", profiles, PlacementPolicies.fixed(), _ -> 0);
+
+        PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
+        assertEquals("b", h.profile(),
+                "fixed placement must fail over from the unreachable default a to the healthy b");
+        assertEquals(1, adapter.spawnCount("a"), "a was tried once and failed");
+        assertEquals(1, adapter.spawnCount("b"), "b was tried once and succeeded");
+    }
+
+    /**
+     * fleetd #315: the same fix — {@code FixedPlacementPolicy} consulting {@code ctx.unreachable()}
+     * — also covers the wiring-bug branch in {@code CompositePeerLauncher.spawn}: a profile that
+     * placement is allowed to choose (it is in the configured candidate list) but that no delegate
+     * declares ({@code byProfile.get(chosen.profile()) == null}). That branch adds the profile to
+     * {@code unreachable} and {@code continue}s without ever calling a launcher, so before this fix
+     * {@code fixed} handed back the same adapterless profile on every remaining attempt too.
+     */
+    @Test
+    void failoverSkipsAConfiguredProfileNoAdapterDeclaresUnderFixedPlacement() {
+        FakeHerdr herdr = new FakeHerdr();
+        // Placement's candidate list has three profiles, in this order (LinkedHashMap preserves it,
+        // and the fixed default resolves to the first — see the `ordered` helper's own javadoc).
+        Map<String, FleetConfig.Profile> profiles = new LinkedHashMap<>();
+        profiles.put("c", stubWorker("c"));
+        profiles.put("a", stubWorker("a"));
+        profiles.put("b", stubWorker("b"));
+        // The adapter only declares a and b — c is a configured profile with no owning adapter,
+        // the "wiring bug" the comment in CompositePeerLauncher.spawn calls out.
+        Map<String, FleetConfig.Profile> adapterProfiles = new LinkedHashMap<>();
+        adapterProfiles.put("a", stubWorker("a"));
+        adapterProfiles.put("b", stubWorker("b"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, adapterProfiles, "a", Set.of());
+        CompositePeerLauncher composite = new CompositePeerLauncher(
+                List.of(adapter), "a", profiles, PlacementPolicies.fixed(), _ -> 0);
+
+        PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
+        assertEquals("a", h.profile(),
+                "c has no adapter, so fixed placement must skip it and land on the next candidate, a");
+        assertEquals(0, adapter.spawnCount("c"), "c is never spawned — no adapter owns it");
+        assertEquals(1, adapter.spawnCount("a"));
+    }
+
     @Test
     void explicitSpawnAtMaxLoadThrowsPlacementExceptionNamingProfileLiveAndCap() {
         FakeHerdr herdr = new FakeHerdr();

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

@@ -957,6 +957,76 @@ class MessageServiceTest {
         assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
     }
 
+    /**
+     * fleetd #307: a worker's {@code fleet_ask} can time out because the primary never answers —
+     * distinct from {@link #aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket}, where the
+     * primary DID answer and only its own bounded wait for the resumed turn expired.
+     * {@code ask()}'s timeout path deliberately forgets the task's {@code turnId} (so
+     * {@code hasAsyncQuestion} stops reporting the target BUSY — see
+     * {@code unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget} above), which used
+     * to also erase the one signal {@code askAnsweredAsyncTasks} needed to recognize the worker's
+     * eventual real {@code fleet_reply}. That reply then had nowhere to land but the inbox, and
+     * {@code fleet_poll{ticket}} stayed PENDING forever — later force-failed with the false reason
+     * "session released before it replied", even though the worker had, in fact, replied.
+     */
+    @Test
+    void aReplyAfterAnAskTimeoutStillCompletesTheAsyncTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks then finishes alone");
+        awaitWaiting();
+        injectDelivery();
+
+        assertEquals(MessageService.AskOutcome.TIMED_OUT,
+                messages.ask(T, "which config?", 200).outcome());
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
+                "only the question wait ended; the delegated turn may still finish");
+
+        // The worker keeps working past the timeout and only now calls fleet_reply — with no live
+        // rendezvous waiter open (ask()'s timeout already closed it) and no new send() having
+        // reopened one for this target.
+        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+        MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+        assertEquals("PR opened: https://example/pulls/42", done.reply(),
+                "fleet_poll{ticket} must return the worker's real reply, not stay pending forever");
+        assertEquals("reply", done.replySource());
+        assertFalse(messages.hasStrandedReply(T),
+                "the reply completed its own ticket directly and never touched the inbox");
+    }
+
+    /**
+     * fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
+     * becomes false the instant it lapses — proven above), so a second, independent delegation can
+     * be dispatched to the same target and itself go on to ask-and-lapse before the first worker's
+     * real reply ever arrives. Two open tasks are then both eligible candidates on one target with
+     * no live waiter to disambiguate them. A reply arriving now must not guess which one it answers
+     * — guessing wrong would hand the lead a plausible-looking answer to a delegation the worker
+     * never touched, worse than a failure because the lead acts on it — so it must fall back to the
+     * inbox exactly as the zero-candidate case does.
+     */
+    @Test
+    void twoAskTimedOutTicketsOnOneTargetFallBackToTheInboxRatherThanGuess() throws Exception {
+        String ticket1 = messages.sendAsync(T, "first task that asks");
+        awaitWaiting();
+        injectDelivery();
+        assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q1?", 200).outcome());
+
+        String ticket2 = messages.sendAsync(T, "second task that asks");
+        awaitWaiting();
+        injectDelivery();
+        assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q2?", 200).outcome());
+
+        assertTrue(messages.reply(T, "which task does this answer?"));
+
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket1).phase(),
+                "an ambiguous reply must not guess ticket1");
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket2).phase(),
+                "an ambiguous reply must not guess ticket2");
+        assertTrue(messages.hasStrandedReply(T));
+        var drained = messages.drainReplies(T);
+        assertEquals(1, drained.size());
+        assertEquals("which task does this answer?", drained.get(0).content());
+    }
+
     @Test
     void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
         String first = messages.sendAsync(T, "first task");

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

@@ -26,7 +26,12 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.LongSupplier;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -81,12 +86,32 @@ class SessionManagerTest {
         private volatile RuntimeException hasUncommittedFailure;
         private volatile RuntimeException snapshotFailure;
         private final java.util.concurrent.atomic.AtomicLong snapshotSeq = new java.util.concurrent.atomic.AtomicLong();
+        /** fleetd #316: successive {@code hasUncommitted} answers, one per call, last one sticky
+         *  once exhausted — models a worktree whose state changes between reads. Empty (the
+         *  default) falls back to the plain {@link #dirty} flag, so every existing test using this
+         *  fake keeps returning one fixed answer. */
+        private final List<Boolean> dirtySequence = new java.util.concurrent.CopyOnWriteArrayList<>();
+        private final java.util.concurrent.atomic.AtomicInteger hasUncommittedCalls =
+                new java.util.concurrent.atomic.AtomicInteger();
 
         RecordingWorktrees dirty(boolean dirty) {
             this.dirty = dirty;
             return this;
         }
 
+        /** fleetd #316: return {@code answers[0]} on the first {@code hasUncommitted} call,
+         *  {@code answers[1]} on the second, and so on; the last element repeats after that. */
+        RecordingWorktrees dirtySequence(boolean... answers) {
+            for (boolean a : answers) {
+                dirtySequence.add(a);
+            }
+            return this;
+        }
+
+        int hasUncommittedCallCount() {
+            return hasUncommittedCalls.get();
+        }
+
         RecordingWorktrees failHasUncommittedWith(RuntimeException e) {
             this.hasUncommittedFailure = e;
             return this;
@@ -122,9 +147,13 @@ class SessionManagerTest {
 
         @Override
         public boolean hasUncommitted(String worktreePath) {
+            int call = hasUncommittedCalls.getAndIncrement();
             if (hasUncommittedFailure != null) {
                 throw hasUncommittedFailure;
             }
+            if (!dirtySequence.isEmpty()) {
+                return dirtySequence.get(Math.min(call, dirtySequence.size() - 1));
+            }
             return dirty;
         }
 
@@ -178,14 +207,19 @@ class SessionManagerTest {
     }
 
     private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
-                                          boolean clearAfterTurn) {
+                                           boolean clearAfterTurn) {
+        return sessionManager(herdr, clock, contextCap, clearAfterTurn, null);
+    }
+
+    private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
+                                           boolean clearAfterTurn, java.util.function.Consumer<MemberSession> hook) {
         FleetConfig.Profile cfg = new FleetConfig.Profile(
                 "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
                 List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
                 "worker: {profile} #{n}", null, null, null);
         ClaudeCodeLauncher workers = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
                 new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
-        return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn);
+        return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn, hook);
     }
 
     @Test
@@ -670,6 +704,24 @@ class SessionManagerTest {
                 "BUSY session remains");
     }
 
+    @Test
+    void reapIdleDoesNotReleaseSessionDeliveredAfterItsEligibilityCheck() {
+        long[] clock = {0};
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager[] manager = new SessionManager[1];
+        SessionManager sessions = sessionManager(herdr, () -> clock[0], 0, false,
+                session -> manager[0].onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId())));
+        manager[0] = sessions;
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+
+        clock[0] = 11;
+        assertEquals(0, sessions.reapIdle(10), "delivery replaces the idle snapshot before release");
+        assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
+                "a just-delivered session stays registered and busy");
+        assertFalse(herdr.called("pane.close"), "the busy session pane is not stopped");
+    }
+
     @Test
     void doneSessionPastIdleTtlIsReaped() {
         long[] clock = {0};
@@ -824,6 +876,186 @@ class SessionManagerTest {
                 .count();
     }
 
+    // --- fleetd #308: a spawn accepted while the shutdown drain is running must not orphan ---
+
+    @Test
+    void acquireRefusesANewSpawnOnceDrainAllHasStarted() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+
+        sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(50)); // empty roster — returns immediately,
+        // but the shutdown guard it flips must stay tripped for the life of the process.
+
+        ShuttingDownException e = assertThrows(ShuttingDownException.class,
+                () -> sessions.acquire("ltms-local", null, "/caller", "term_primary"),
+                "a spawn requested after the drain has begun must be refused loudly (invariant 3), "
+                        + "not silently registered into a registry the drain will never revisit");
+        assertNotNull(e.getMessage());
+        assertFalse(e.getMessage().isBlank(), "the refusal must say why, not just that it failed");
+        assertTrue(sessions.roster().isEmpty(), "the refused spawn must never reach the registry");
+    }
+
+    /**
+     * fleetd #308: the guard above closes most of the shutdown-race window, but it cannot close
+     * all of it — a caller that already passed the {@code draining} check before {@code drainAll}
+     * flips it can still be mid-{@code launcher.spawn()} (a real herdr round trip, not
+     * instantaneous) when {@code drainAll} takes its registry snapshot. This test forces exactly
+     * that interleaving with a launcher double that blocks the second {@code spawn()} call and the
+     * first {@code stop()} call until released, then proves the post-loop sweep in {@code
+     * drainAll} still finds and tears down the straggler that lands in the registry afterward.
+     */
+    @Test
+    void drainAllSweepsAStragglerThatRegisteredAfterTheInitialSnapshot() throws Exception {
+        FakeHerdr herdr = new FakeHerdr();
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
+                List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher delegate = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
+        RaceLauncher race = new RaceLauncher(delegate);
+        SessionManager sessions = new SessionManager(race);
+
+        // Registered normally, before the drain starts — the first spawn call, never blocked.
+        MemberSession ready = sessions.acquire("ltms-local", "/ready", "/caller", "ownerR");
+
+        ExecutorService exec = Executors.newFixedThreadPool(2);
+        try {
+            // The straggler's acquire() reads `draining == false` (checked before this call ever
+            // touches the launcher) and then blocks inside its own spawn() — the second spawn call.
+            Future<MemberSession> straggler = exec.submit(() ->
+                    sessions.acquire("ltms-local", "/late", "/caller", "ownerLate"));
+
+            assertTrue(race.enteredSecondSpawn.await(5, TimeUnit.SECONDS),
+                    "the straggler must have passed the shutdown guard and reached spawn() before "
+                            + "drainAll ever runs");
+            assertEquals(1, sessions.roster().size(),
+                    "the straggler is still inside spawn() — not registered yet");
+
+            // drainAll flips `draining`, snapshots the registry (only `ready` is in it), and starts
+            // releasing that snapshot — its first release() call stops `ready`'s pane, which this
+            // launcher double blocks on so the interleaving below is deterministic, not a timing bet.
+            Future<?> drain = exec.submit(() -> sessions.drainAll(TimeUnit.SECONDS.toNanos(5)));
+
+            assertTrue(race.enteredFirstStop.await(5, TimeUnit.SECONDS),
+                    "drainAll must be stopping the ready session's pane — proof its initial "
+                            + "registry snapshot has already been taken");
+
+            // Only now does the straggler's spawn complete and register — strictly after the
+            // snapshot drainAll's main pass is working from.
+            race.releaseSecondSpawn.countDown();
+            MemberSession registered = straggler.get(5, TimeUnit.SECONDS);
+
+            // Let drainAll finish releasing `ready`; it then re-checks the registry and must find
+            // (and drain) the straggler that just landed in it.
+            race.releaseFirstStop.countDown();
+            drain.get(5, TimeUnit.SECONDS);
+
+            assertTrue(sessions.roster().isEmpty(),
+                    "the post-loop sweep must drain the straggler too, not just the initial snapshot");
+            assertNotNull(registered.paneId());
+            long paneCloseCalls = herdr.calls.stream().filter(c -> "pane.close".equals(c.method())).count();
+            assertEquals(2, paneCloseCalls,
+                    "both ready's pane AND the straggler's pane must actually be stopped — a pane "
+                            + "left running is exactly the orphan this ticket is about");
+        } finally {
+            exec.shutdownNow();
+        }
+    }
+
+    /**
+     * Delegates every call while blocking the SECOND {@code spawn()} call and the FIRST
+     * {@code stop()} call until the test releases them — used to force the fleetd #308 race
+     * deterministically instead of betting on real thread-scheduling timing.
+     */
+    private static final class RaceLauncher implements PeerLauncher {
+        private final PeerLauncher delegate;
+        private final AtomicInteger spawnCalls = new AtomicInteger();
+        private final AtomicInteger stopCalls = new AtomicInteger();
+        final CountDownLatch enteredSecondSpawn = new CountDownLatch(1);
+        final CountDownLatch releaseSecondSpawn = new CountDownLatch(1);
+        final CountDownLatch enteredFirstStop = new CountDownLatch(1);
+        final CountDownLatch releaseFirstStop = new CountDownLatch(1);
+
+        RaceLauncher(PeerLauncher delegate) {
+            this.delegate = delegate;
+        }
+
+        private static void awaitOrFail(CountDownLatch latch) {
+            try {
+                if (!latch.await(5, TimeUnit.SECONDS)) {
+                    throw new AssertionError("RaceLauncher latch timed out");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("RaceLauncher latch interrupted", e);
+            }
+        }
+
+        @Override
+        public Set<Capability> capabilities() {
+            return delegate.capabilities();
+        }
+
+        @Override
+        public Set<Capability> capabilitiesFor(String profileName) {
+            return delegate.capabilitiesFor(profileName);
+        }
+
+        @Override
+        public PeerHandle spawn(SpawnRequest req) {
+            if (spawnCalls.incrementAndGet() == 2) {
+                enteredSecondSpawn.countDown();
+                awaitOrFail(releaseSecondSpawn);
+            }
+            return delegate.spawn(req);
+        }
+
+        @Override
+        public Set<String> profiles() {
+            return delegate.profiles();
+        }
+
+        @Override
+        public String defaultProfile() {
+            return delegate.defaultProfile();
+        }
+
+        @Override
+        public String effectiveCwd(SpawnRequest req) {
+            return delegate.effectiveCwd(req);
+        }
+
+        @Override
+        public List<String> parityOverlay(String profileName) {
+            return delegate.parityOverlay(profileName);
+        }
+
+        @Override
+        public List<?> list() {
+            return delegate.list();
+        }
+
+        @Override
+        public int reapOrphanWorkers() {
+            return delegate.reapOrphanWorkers();
+        }
+
+        @Override
+        public void stop(String id) {
+            if (stopCalls.incrementAndGet() == 1) {
+                enteredFirstStop.countDown();
+                awaitOrFail(releaseFirstStop);
+            }
+            delegate.stop(id);
+        }
+
+        @Override
+        public boolean clearContext(String id) {
+            return delegate.clearContext(id);
+        }
+    }
+
     private static List<String> promptTexts(FakeHerdr herdr) {
         return herdr.calls.stream()
                 .filter(c -> "agent.prompt".equals(c.method()))
@@ -999,6 +1231,80 @@ class SessionManagerTest {
                         + "dirty check threw");
     }
 
+    // --- fleetd #316: the dirty check must be re-taken after the worker is stopped, not trusted
+    // stale from before it ------------------------------------------------------------------------
+
+    @Test
+    void releaseDoesNotRemoveAWorktreeThatBecameDirtyBetweenTheFirstCheckAndRemoval() {
+        // Models the exact race #316 reports: hasUncommitted answers clean while the worker is
+        // still running (call 1), the worker then writes new work, and by the time release is
+        // about to force-remove the worktree a second read (call 2) would see it as dirty. Without
+        // the fix this test fails: release() never re-reads and force-removes the worktree anyway.
+        FakeHerdr herdr = new FakeHerdr();
+        RecordingWorktrees worktrees = new RecordingWorktrees().dirtySequence(false, true);
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-316a", null));
+
+        sessions.release(s.paneId());
+
+        assertTrue(worktrees.removeCalls().isEmpty(),
+                "a worktree that turned dirty between the pre-stop read and removal must be preserved");
+        assertEquals(2, worktrees.hasUncommittedCallCount(),
+                "the fix re-reads hasUncommitted exactly once more, immediately before removal");
+    }
+
+    @Test
+    void releaseSnapshotsWorkFoundOnlyByTheLateRecheck() {
+        // #316's second half: the pre-stop dirty=false means trySnapshot never ran for this
+        // session, so the late-discovered work would otherwise have no refs/wip/* copy at all —
+        // only the on-disk preserve. The re-check path must snapshot it too.
+        FakeHerdr herdr = new FakeHerdr();
+        RecordingWorktrees worktrees = new RecordingWorktrees().dirtySequence(false, true);
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-316b", null));
+
+        sessions.release(s.paneId());
+
+        assertEquals(java.util.List.of(s.worktree()), worktrees.snapshotCalls(),
+                "the newly-dirty worktree is snapshotted even though the pre-stop check saw it clean");
+    }
+
+    @Test
+    void releaseStillRemovesAWorktreeThatStaysCleanOnTheLateRecheck() {
+        // The ordinary, non-racing case: nothing else changes behaviour when the second read
+        // agrees with the first.
+        FakeHerdr herdr = new FakeHerdr();
+        RecordingWorktrees worktrees = new RecordingWorktrees().dirty(false);
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-316c", null));
+
+        sessions.release(s.paneId());
+
+        assertEquals(java.util.List.of(s.worktree()), worktrees.removeCalls(),
+                "a worktree that is still clean on the late recheck is removed as before");
+    }
+
+    @Test
+    void releaseNeverReChecksAWorktreeAlreadyPreservedByTheFirstDirtyCheck() {
+        // Invariant 4 from #316: no second unconditional git status. A release that already
+        // decided to preserve (the ordinary CB-576 dirty path) must not pay for a second read.
+        FakeHerdr herdr = new FakeHerdr();
+        RecordingWorktrees worktrees = new RecordingWorktrees().dirty(true);
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-316d", null));
+
+        sessions.release(s.paneId());
+
+        assertEquals(1, worktrees.hasUncommittedCallCount(),
+                "a release that already preserves on the first read must not re-check before "
+                        + "skipping the removal it was never going to do");
+        assertTrue(worktrees.removeCalls().isEmpty());
+    }
+
     /**
      * fleetd #283 defect 1 changed this test's own premise, so its assertions are updated along
      * with the production fix. Before #283, the middle session's worktree-removal failure escaped