fleetd #702 : pin the architect-demotion window and split a thin test

Add a CallerResolverTest case proving a releasing architect resolves as WORKER while still inside releaseRemoved's teardown window, with a control resolve outside the window that must stay ARCHITECT so the in-window assertion cannot pass against a slot that was never bound. Split spawnedMemberRoleSurvivesAnOverlappingReleaseThatUnmarksEarly into two SessionManagerTest cases, each reaching Releasing.enter/leave directly through reflection so a mutation to one invariant (the depth count in leave, or enter's prior-terminal preservation) can only fail its own test.
fleetd #702 : mark a pane as mid-teardown so a caller still resolves as a member
2026-10-04 10:04:00 +02:00 · 2026-10-04 09:27:09 +02:00
14 changed files with 626 additions and 792 deletions
@@ -51,7 +51,6 @@ import dev.ltms.fleet.power.CaffeinateSleepAssertionMechanism;
 import dev.ltms.fleet.power.IdleSleepGuard;
 import dev.ltms.fleet.rest.FleetApp;
 import dev.ltms.fleet.session.GitWorktrees;
-import dev.ltms.fleet.session.MemberSession;
 import dev.ltms.fleet.session.SessionManager;
 import dev.ltms.fleet.session.SessionReaper;
 import io.javalin.Javalin;
@@ -480,13 +479,9 @@ final class FleetdAssembly {
                new PaneLocator(herdr, memberHerdr), new LsofPeerPidLookup(), new LsofProcessCwdLookup());

        // fleetd #669 Unit D: a live spawned member resolves as its own role, whatever a tab map
-        // says about the same terminal — read from the roster meant for a hot path (SessionManager
-        // javadoc), never rosterResolved(), since resolve() runs on every request.
-        Function<String, MemberRole> spawnedMemberRole = terminal -> sessions.roster().stream()
-                .filter(s -> terminal.equals(s.terminalId()))
-                .map(MemberSession::role)
-                .findFirst()
-                .orElse(null);
+        // says about the same terminal. fleetd #702: SessionManager.spawnedMemberRole also answers
+        // for a pane mid-teardown, not only one still in the registry — see its javadoc.
+        Function<String, MemberRole> spawnedMemberRole = sessions::spawnedMemberRole;

        // CB-501: one resolver behind both entry paths. Worker identity still comes from the
        // connection and is never token-gated, so enabling token mode cannot lock the fleet out.
@@ -480,7 +480,7 @@ public final class FleetMcp {
                        // Answering a worker's fleet_ask (CB-205): resolve its blocked question and
                        // block for the worker's reply as it resumes the same turn. This is the same
                        // delegation, so ownership is left untouched (CB-548) — never re-recorded.
-                        return answer(messages, turnId, content, timeoutMs(a), caller);
+                        return answer(messages, turnId, content, timeoutMs(a));
                    }
                    // CB-548: delegator ownership (which lead's reply nudge this worker routes to,
                    // CB-532) is recorded only once the send is ACCEPTED — MessageService has won the
@@ -491,7 +491,7 @@ public final class FleetMcp {
                    // wait defaults to true (block for the reply); wait:false is fire-and-poll.
                    return Boolean.FALSE.equals(a.get("wait"))
                            ? sendAsync(messages, target, content, onAccepted, workers.profiles(), caller)
-                            : send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles(), caller);
+                            : send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles());
                };
        // fleet_reply's identity is the CONNECTION, never an argument — so the authz check
        // is "is this caller a worker at all", and it can only ever reply as itself.
@@ -787,11 +787,7 @@ public final class FleetMcp {
        return caller.isPrimary() || caller.isArchitect();
    }

-    /**
-     * The terminal of the caller on this call's connection, or {@code null} when that caller carries
-     * no terminal, which is only the unnamed primary. A named lead, an architect and a worker each
-     * carry one.
-     */
+    /** The worker identity resolved from this call's connection, or {@code null} if the primary. */
    private static String callerTerminal(McpSyncServerExchange exchange) {
        Object v = exchange.transportContext().get(CALLER_TERMINAL);
        String s = v == null ? null : v.toString();
@@ -908,8 +904,7 @@ public final class FleetMcp {
     * a BUSY interloper never claims a turn it did not win. {@code null} disables recording.
     */
    static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content,
-                                         Long timeoutMs, Runnable onAccepted, Set<String> profiles,
-                                         String callerTerminal) {
+                                         Long timeoutMs, Runnable onAccepted, Set<String> profiles) {
        if (isBlank(sessionId) || isBlank(content)) {
            return error("sessionId and content are required");
        }
@@ -919,7 +914,7 @@ public final class FleetMcp {
        }
        long timeout = clamp(timeoutMs == null ? DEFAULT_TIMEOUT_MS : timeoutMs);
        try {
-            return formatReply(messages.send(sessionId, content, timeout, onAccepted, callerTerminal), timeout);
+            return formatReply(messages.send(sessionId, content, timeout, onAccepted), timeout);
        } catch (HerdrException e) {
            return error("herdr error contacting session " + sessionId + ": " + e.getMessage());
        }
@@ -929,15 +924,13 @@ public final class FleetMcp {
     * {@code fleet_send} carrying a {@code turnId}: the primary's answer to a worker's
     * {@code fleet_ask} (CB-205). Resolves the worker's blocked question and blocks for its reply as
     * it resumes the same turn — surfaced to the primary identically to a normal send.
-     * {@code callerTerminal} must match the turn's recorded owner or this is refused.
     */
-    static McpSchema.CallToolResult answer(MessageService messages, String turnId, String content, Long timeoutMs,
-                                            String callerTerminal) {
+    static McpSchema.CallToolResult answer(MessageService messages, String turnId, String content, Long timeoutMs) {
        if (isBlank(turnId) || isBlank(content)) {
            return error("turnId and content are required to answer a worker's question");
        }
        long timeout = clamp(timeoutMs == null ? DEFAULT_TIMEOUT_MS : timeoutMs);
-        return formatReply(messages.answer(turnId, content, timeout, callerTerminal), timeout);
+        return formatReply(messages.answer(turnId, content, timeout), timeout);
    }

    /**
@@ -986,8 +979,6 @@ public final class FleetMcp {
                    + "\" and content set to your answer; the worker resumes the same turn.");
            case STALE_TURN -> error("that question is no longer open — it timed out or was already "
                    + "answered (turnId stale)");
-            case NOT_TURN_OWNER -> error("this turn belongs to a different delegation — only the caller "
-                    + "that opened it may answer it");
            case TIMED_OUT_WORKING, TIMED_OUT_QUEUED, BUSY -> text("[no reply within " + timeout + "ms — worker "
                    + r.outcome().name().toLowerCase().replace("timed_out_", "") + "; retry or poll status]");
            // fleetd #571: delivery is unknown here — agent.prompt pastes and submits in one call,
@@ -87,7 +87,7 @@ public final class MessageService {
        /**
         * The worker paused mid-turn to ask the primary a question (CB-205); {@code text} is the
         * question and {@code turnId} correlates the answer. Not terminal — the primary answers with
-         * {@link #answer(String, String, long, String)} and the turn resumes.
+         * {@link #answer(String, String, long)} and the turn resumes.
         */
        QUESTION,
        /** Timed out after the message was delivered — the worker is still working. */
@@ -120,16 +120,10 @@ public final class MessageService {
        /** Another send to this session was in flight for the whole window. */
        BUSY,
        /**
-         * An answer ({@link #answer(String, String, long, String)}) referenced a {@code turnId}
-         * that is no longer open — the worker's {@code fleet_ask} already timed out or was answered.
+         * An answer ({@link #answer(String, String, long)}) referenced a {@code turnId} that is no
+         * longer open — the worker's {@code fleet_ask} already timed out or was answered.
         */
-        STALE_TURN,
-        /**
-         * An answer ({@link #answer(String, String, long, String)}) named a {@code turnId} that is
-         * still open, but the answering caller is not the caller whose accepted delegation opened
-         * it. Distinct from {@link #STALE_TURN} so a refusal is never reported as a lapsed turn.
-         */
-        NOT_TURN_OWNER
+        STALE_TURN
    }

    /**
@@ -139,7 +133,7 @@ public final class MessageService {
     *                {@link Outcome#COMPLETED_UNREPLIED}), or the question for {@link Outcome#QUESTION},
     *                else {@code null}
     * @param turnId  correlation id for a {@link Outcome#QUESTION} (answered via
-     *                {@link #answer(String, String, long, String)}), else {@code null}
+     *                {@link #answer(String, String, long)}), else {@code null}
     */
    public record Reply(Outcome outcome, String text, String turnId) {
        /** A reply with no correlation id (the common terminal outcomes). */
@@ -671,7 +665,7 @@ public final class MessageService {
            case TIMED_OUT_WORKING, TIMED_OUT_QUEUED, TIMED_OUT_UNCONFIRMED, BUSY -> "timeout";
            case WORKER_FAILED -> "failed";
            case BACKEND_EXHAUSTED -> "backend_exhausted";
-            case STALE_TURN, QUESTION, NOT_TURN_OWNER -> null; // not a completed delegation
+            case STALE_TURN, QUESTION -> null; // not a completed delegation
        };
    }

@@ -930,17 +924,14 @@ public final class MessageService {

    /**
     * Deliver {@code content} to {@code target} (a herdr {@code terminal_id}) and block until the
-     * worker replies via {@link Rendezvous} or {@code timeoutMillis} elapses. {@code callerTerminal}
-     * is the terminal of the caller making this call — {@code null} for the unnamed primary — and is
-     * recorded as the turn's owner, the only caller {@link #answer(String, String, long, String)} will
-     * later accept an answer from if the worker pauses mid-turn to ask.
+     * worker replies via {@link Rendezvous} or {@code timeoutMillis} elapses.
     */
-    public Reply send(String target, String content, long timeoutMillis, String callerTerminal) {
-        return send(target, content, timeoutMillis, null, callerTerminal);
+    public Reply send(String target, String content, long timeoutMillis) {
+        return send(target, content, timeoutMillis, null);
    }

    /**
-     * As {@link #send(String, String, long, String)}, but with an accepted-delivery hook.
+     * As {@link #send(String, String, long)}, but with an accepted-delivery hook.
     *
     * <p>{@code onAccepted} is invoked exactly once, once this send has won {@code target}'s send
     * lock and so become the <em>accepted target turn</em> — it runs <em>before</em> delivery is
@@ -951,13 +942,12 @@ public final class MessageService {
     * acceptance means a concurrent sender that times out {@code BUSY} can never steal ownership it
     * never earned. {@code null} disables the hook.
     */
-    public Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, String callerTerminal) {
-        return send(target, content, timeoutMillis, onAccepted, null, callerTerminal);
+    public Reply send(String target, String content, long timeoutMillis, Runnable onAccepted) {
+        return send(target, content, timeoutMillis, onAccepted, null);
    }

    /** Run a send, optionally stopping an async task that teardown already failed before acceptance. */
-    private Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, Task task,
-            String callerTerminal) {
+    private Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, Task task) {
        long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L;
        ReentrantLock lock = sessionLocks.computeIfAbsent(target, _ -> new ReentrantLock());

@@ -977,7 +967,7 @@ public final class MessageService {
            // open race). Opening first also means a throwing onAccepted (fired before enqueue) or an
            // enqueue failure is safely closed by the finally below: nothing is left queued, and the
            // failed send leaves no stale waiter behind.
-            CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target, Rendezvous.Owner.of(callerTerminal));
+            CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
            // CB-640: this send now owns target's delivery, so any earlier stranded-reply or
            // still-queued fact no longer describes the live state — clear both rather than let
            // them outlive the send that supersedes them.
@@ -1156,30 +1146,19 @@ public final class MessageService {
     * mid-turn (already picked up), so the answer flows back through its own open {@code fleet_ask}
     * call, not a new status-gated delivery. The forward waiter is opened <em>before</em> the worker
     * is unblocked so a reply that lands the instant it resumes is not lost.
-     *
-     * <p>{@code callerTerminal} is the terminal of the caller making this call — {@code null} for
-     * the unnamed primary. It is checked against the turn's recorded owner (the caller whose
-     * accepted delegation opened it, see {@link #send(String, String, long, String)} and
-     * {@link #sendAsync(String, String, Runnable, String)}) before anything else runs: a mismatch,
-     * including a turn with no owner on record at all, returns {@link Outcome#NOT_TURN_OWNER}
-     * without touching the rendezvous, the session lock, or any async task bookkeeping.
     */
-    public Reply answer(String turnId, String content, long timeoutMillis, String callerTerminal) {
+    public Reply answer(String turnId, String content, long timeoutMillis) {
        String workerSession = rendezvous.askSession(turnId);
        if (workerSession == null) {
            return new Reply(Outcome.STALE_TURN, null); // the ask lapsed (timed out or already answered)
        }
-        Rendezvous.Owner owner = rendezvous.askOwner(turnId);
-        if (!Rendezvous.Owner.permits(owner, callerTerminal)) {
-            return new Reply(Outcome.NOT_TURN_OWNER, null);
-        }
        long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L;
        ReentrantLock lock = sessionLocks.computeIfAbsent(workerSession, _ -> new ReentrantLock());
        if (!tryLock(lock, remainingMillis(deadlineNanos))) {
            return new Reply(Outcome.BUSY, null);
        }
        try {
-            CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(workerSession, owner);
+            CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(workerSession);
            // fleetd #575: this try used to open below, AFTER the Task lookup/registration and the
            // STALE_TURN early return that follows it — so that return was covered only by a
            // hand-rolled copy of the finally's own cleanup pair, not the finally itself. Widening the
@@ -1353,7 +1332,7 @@ public final class MessageService {
        }
        asyncExecutor.submit(() -> {
            try {
-                Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task, creatorTerminal);
+                Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task);
                if (result.outcome() == Outcome.QUESTION) {
                    // Keep the accepted owner until answer() finishes it. markAsyncQuestion may run
                    // just after resolveQuestion wakes this thread.
@@ -60,7 +60,7 @@ public final class Rendezvous {
    }

    /** A worker's open mid-turn question: the worker session it belongs to and the answer future. */
-    private record AskWaiter(String session, CompletableFuture<String> answer, Owner owner) {
+    private record AskWaiter(String session, CompletableFuture<String> answer) {
    }

    /**
@@ -70,33 +70,7 @@ public final class Rendezvous {
    public record AskTicket(String turnId, CompletableFuture<String> answer, boolean fresh) {
    }

-    /**
-     * The caller whose accepted delegation opened a turn — the only caller allowed to answer it.
-     * A {@code null} terminal means the unnamed primary, an authenticated caller with no pane.
-     */
-    public record Owner(String terminal) {
-        public static final Owner UNNAMED_PRIMARY = new Owner(null);
-
-        public static Owner of(String terminal) {
-            return terminal == null ? UNNAMED_PRIMARY : new Owner(terminal);
-        }
-
-        /**
-         * Whether {@code callerTerminal} matches {@code owner}. A {@code null} owner means no
-         * owner was ever recorded, and that state matches no caller, not even one whose own
-         * terminal is {@code null} — "no record" and "recorded as the unnamed primary" are
-         * different states.
-         */
-        public static boolean permits(Owner owner, String callerTerminal) {
-            return owner != null && java.util.Objects.equals(owner.terminal(), callerTerminal);
-        }
-    }
-
-    /** A registered forward waiter together with the owner its delegation was opened under. */
-    private record ForwardWaiter(Owner owner, CompletableFuture<Resolution> future) {
-    }
-
-    private final ConcurrentHashMap<String, ForwardWaiter> waiters = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, CompletableFuture<Resolution>> waiters = new ConcurrentHashMap<>();

    /** Reverse rendezvous (CB-205): worker questions awaiting the primary's answer, keyed by {@code turnId}. */
    private final ConcurrentHashMap<String, AskWaiter> asks = new ConcurrentHashMap<>();
@@ -115,17 +89,8 @@ public final class Rendezvous {
     * code a double open is impossible; this is a tripwire for the day that no longer holds.
     */
    public CompletableFuture<Resolution> open(String session) {
-        return open(session, null);
-    }
-
-    /**
-     * Same as {@link #open(String)}, additionally recording {@code owner} as the caller whose
-     * delegation opened this waiter. A {@code null} owner records no owner at all — the
-     * fail-closed default {@link Owner#permits} refuses to everyone.
-     */
-    public CompletableFuture<Resolution> open(String session, Owner owner) {
        CompletableFuture<Resolution> waiter = new CompletableFuture<>();
-        ForwardWaiter existing = waiters.putIfAbsent(session, new ForwardWaiter(owner, waiter));
+        CompletableFuture<Resolution> existing = waiters.putIfAbsent(session, waiter);
        if (existing != null) {
            throw new IllegalStateException(
                    "rendezvous double-open for session " + session + " — a waiter is already registered");
@@ -140,13 +105,7 @@ public final class Rendezvous {
     * successful {@code open} after a finished turn requires this close to have happened first).
     */
    public void close(String session, CompletableFuture<Resolution> waiter) {
-        waiters.computeIfPresent(session, (s, w) -> w.future() == waiter ? null : w);
-    }
-
-    /** The owner recorded for {@code session}'s open waiter, or {@code null} if none is open. */
-    public Owner ownerOf(String session) {
-        ForwardWaiter w = waiters.get(session);
-        return w == null ? null : w.owner();
+        waiters.remove(session, waiter);
    }

    /** Whether a send is currently awaiting a resolution for {@code session}. */
@@ -160,8 +119,7 @@ public final class Rendezvous {
     * send (see the CB-116 note above) rather than whichever send happens to be waiting when they fire.
     */
    public CompletableFuture<Resolution> currentWaiter(String session) {
-        ForwardWaiter w = waiters.get(session);
-        return w == null ? null : w.future();
+        return waiters.get(session);
    }

    /**
@@ -189,7 +147,7 @@ public final class Rendezvous {
            String turnId = openAsksBySession.computeIfAbsent(session, _ -> {
                String newTurnId = session + "#" + askSeq.incrementAndGet();
                CompletableFuture<String> answer = new CompletableFuture<>();
-                AskWaiter waiter = new AskWaiter(session, answer, ownerOf(session));
+                AskWaiter waiter = new AskWaiter(session, answer);
                asks.put(newTurnId, waiter);
                minted[0] = waiter;
                return newTurnId;
@@ -225,16 +183,6 @@ public final class Rendezvous {
        return w == null ? null : w.session();
    }

-    /**
-     * The owner recorded for {@code turnId} when its ask turn was freshly opened — the caller
-     * whose delegation {@link #answerAsk} must match. {@code null} if {@code turnId} is unknown or
-     * lapsed, or if the ask opened with no forward waiter owner on record.
-     */
-    public Owner askOwner(String turnId) {
-        AskWaiter w = asks.get(turnId);
-        return w == null ? null : w.owner();
-    }
-
    /**
     * Resolve a worker's blocked {@code fleet_ask} with the primary's {@code answer}, unblocking it
     * to resume its turn.
@@ -297,7 +245,7 @@ public final class Rendezvous {
    }

    private boolean complete(String session, Resolution resolution) {
-        ForwardWaiter waiter = waiters.get(session);
-        return waiter != null && waiter.future().complete(resolution);
+        CompletableFuture<Resolution> waiter = waiters.get(session);
+        return waiter != null && waiter.complete(resolution);
    }
 }
@@ -663,8 +663,6 @@ public final class FleetApp {
        if (!allow(ctx, routeAction("POST /sessions/{id}/message"), id)) {
            return;
        }
-        Principal caller = ctx.attribute(CALLER);
-        String callerTerminal = caller == null ? null : caller.terminal();
        JsonNode body;
        try {
            body = mapper.readTree(ctx.body());
@@ -690,19 +688,20 @@ public final class FleetApp {

        // Answering a worker's fleet_ask (CB-205): always blocks, and derives the worker from turnId.
        if (turnId != null && !turnId.isBlank()) {
-            writeReply(ctx, id, messages.answer(turnId, content, timeout, callerTerminal), timeout);
+            writeReply(ctx, id, messages.answer(turnId, content, timeout), timeout);
            return;
        }

        if (!wait) {
            // Fire-and-poll (CB-107): return a ticket immediately; the caller polls GET /tasks/{ticket}.
-            String ticket = messages.sendAsync(id, content, null, callerTerminal);
+            Principal caller = ctx.attribute(CALLER);
+            String ticket = messages.sendAsync(id, content, null, caller == null ? null : caller.terminal());
            ctx.status(202).json(Map.of("sessionId", id, "ticket", ticket, "status", "accepted"));
            return;
        }

        try {
-            writeReply(ctx, id, messages.send(id, content, timeout, callerTerminal), timeout);
+            writeReply(ctx, id, messages.send(id, content, timeout), timeout);
        } catch (HerdrException e) {
            herdrError(ctx, e);
        }
@@ -721,10 +720,6 @@ public final class FleetApp {
            case STALE_TURN -> ctx.status(409).json(Map.of(
                    "sessionId", id, "error", "stale_turn",
                    "detail", "that question is no longer open (timed out or already answered)"));
-            case NOT_TURN_OWNER -> ctx.status(403).json(Map.of(
-                    "sessionId", id, "error", "not_turn_owner",
-                    "detail", "this turn belongs to a different delegation — only the caller that "
-                            + "opened it may answer it"));
            case REPLIED, COMPLETED_UNREPLIED -> {
                // replySource distinguishes a structured fleet_reply from the CB-106 completion
                // fallback (a scrape of the worker's transcript when it finished without replying).
@@ -739,10 +734,9 @@ public final class FleetApp {
                    // silent fall-through. That is exactly the bug this ticket exists to fix:
                    // `default -> "done"` used to sit here and would have told a REST caller the
                    // delegation completed for TIMED_OUT_UNCONFIRMED, the one outcome where delivery
-                    // is unknown. REPLIED, COMPLETED_UNREPLIED, QUESTION, STALE_TURN and
-                    // NOT_TURN_OWNER can never actually reach this inner switch — the outer switch
-                    // above always dispatches them first — but they still need an arm to keep this
-                    // switch exhaustive.
+                    // is unknown. REPLIED, COMPLETED_UNREPLIED, QUESTION and STALE_TURN can never
+                    // actually reach this inner switch — the outer switch above always dispatches
+                    // them first — but they still need an arm to keep this switch exhaustive.
                    "status", switch (reply.outcome()) {
                        case TIMED_OUT_WORKING -> "working";
                        case TIMED_OUT_QUEUED -> "queued";
@@ -752,7 +746,7 @@ public final class FleetApp {
                        case BUSY -> "busy";
                        case WORKER_FAILED -> "failed";
                        case BACKEND_EXHAUSTED -> "backend_exhausted";
-                        case REPLIED, COMPLETED_UNREPLIED, QUESTION, STALE_TURN, NOT_TURN_OWNER -> "done"; // unreachable
+                        case REPLIED, COMPLETED_UNREPLIED, QUESTION, STALE_TURN -> "done"; // unreachable
                    },
                    "detail", reply.outcome() == MessageService.Outcome.TIMED_OUT_UNCONFIRMED
                            ? "no reply within " + timeout + "ms; delivery is unconfirmed — the "
@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
+import java.util.function.Supplier;

 /**
 * Authoritative in-daemon registry of the worker sessions this {@code fleetd} process spawned.
@@ -57,6 +58,18 @@ public final class SessionManager implements TurnListener {
     * Populated on every spawn path, removed on {@link #release}.
     */
    private final ConcurrentHashMap<String /*paneId*/, PeerHandle> handles = new ConcurrentHashMap<>();
+    /**
+     * fleetd #702: a pane mid-teardown, keyed by paneId, held from just before its registry entry
+     * is removed until {@link #releaseRemoved} finishes. {@link #spawnedMemberRole} consults this
+     * alongside the registry, so a caller resolving the pane's terminal during that window still
+     * sees a live member and never falls through to a tab map.
+     *
+     * <p>Depth-counted rather than a plain set: two threads can be tearing down the same pane at
+     * once (the CAS in {@link #releaseIfCurrent} exists for exactly that race), and with a set the
+     * loser's {@code finally} would unmark the pane while the winner is still mid-teardown,
+     * reopening the window this exists to close.
+     */
+    private final ConcurrentHashMap<String /*paneId*/, Releasing> releasing = new ConcurrentHashMap<>();
    private final MemberPresence presence;
    private final SecureRandom nonceRandom = new SecureRandom();
    private final AtomicLong nonceSeq = new AtomicLong();
@@ -303,9 +316,11 @@ public final class SessionManager implements TurnListener {
     * is a logged path an operator can reclaim, the cost of a deleted one is unrecoverable work.
     */
    private MemberSession release(String paneId, ReleaseCause cause) {
-        MemberSession removed = registry.remove(paneId);
-        releaseRemoved(paneId, removed, handles.remove(paneId), cause);
-        return removed;
+        return releaseWindow(paneId, registry.get(paneId), () -> {
+            MemberSession removed = registry.remove(paneId);
+            releaseRemoved(paneId, removed, handles.remove(paneId), cause);
+            return removed;
+        });
    }

    /**
@@ -314,16 +329,85 @@ public final class SessionManager implements TurnListener {
     * 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;
+        return releaseWindow(expected.paneId(), expected, () -> {
+            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;
+        });
+    }
+
+    /**
+     * fleetd #702: mark {@code paneId} as mid-teardown — using {@code known}'s terminal/role when
+     * it is available — for the whole of {@code teardown}, which removes the registry entry and
+     * then runs {@link #releaseRemoved}. Shared by both registry-removal sites ({@link #release}'s
+     * unconditional remove and {@link #releaseIfCurrent}'s CAS remove) so neither can leave the
+     * other's window unmarked.
+     *
+     * <p>The mark is written before {@code teardown} runs — so it covers the removal itself, not
+     * only what comes after it — and cleared in a {@code finally}, so an unchecked throw out of
+     * {@code teardown} (including one from {@link PeerLauncher#stop}, which declares nothing) can
+     * never leave the pane marked for the rest of the daemon's life.
+     */
+    private <T> T releaseWindow(String paneId, MemberSession known, Supplier<T> teardown) {
+        releasing.compute(paneId, (_, prior) -> Releasing.enter(prior, known));
+        try {
+            return teardown.get();
+        } finally {
+            releasing.compute(paneId, (_, prior) -> prior == null ? null : prior.leave());
        }
-        releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
-        return true;
+    }
+
+    /**
+     * Depth count plus the terminal/role a mid-teardown pane belongs to, for
+     * {@link #spawnedMemberRole}. The terminal/role come from whichever call into
+     * {@link #releaseWindow} first knew them — a losing {@link #releaseIfCurrent} CAS has no
+     * {@code known} session of its own, so it must not blank out what the winner already recorded.
+     */
+    private record Releasing(int depth, String terminalId, MemberRole role) {
+        private static Releasing enter(Releasing prior, MemberSession known) {
+            int depth = (prior == null ? 0 : prior.depth()) + 1;
+            String terminalId = known != null ? known.terminalId() : prior == null ? null : prior.terminalId();
+            MemberRole role = known != null ? known.role() : prior == null ? null : prior.role();
+            return new Releasing(depth, terminalId, role);
+        }
+
+        private Releasing leave() {
+            return depth <= 1 ? null : new Releasing(depth - 1, terminalId, role);
+        }
+    }
+
+    /**
+     * The role of the live spawned member occupying {@code terminal} — whether it is currently in
+     * the registry, or mid-teardown between {@link #release} removing its registry entry and
+     * {@link #releaseRemoved} actually stopping its pane (fleetd #702). {@code null} for a terminal
+     * that is neither: this method is the one reader a caller resolver consults before any tab
+     * map, so a live or releasing member's identity never falls back to a tab label.
+     *
+     * <p>Checks the registry directly via {@link #findByTerminal} rather than {@link #roster()},
+     * so this hot-path lookup (consulted on every resolve) never pays for a list copy or a stream.
+     */
+    public MemberRole spawnedMemberRole(String terminal) {
+        MemberSession session = findByTerminal(terminal);
+        if (session != null) {
+            return session.role();
+        }
+        if (terminal == null) {
+            return null;
+        }
+        for (Releasing r : releasing.values()) {
+            if (terminal.equals(r.terminalId())) {
+                return r.role();
+            }
+        }
+        return null;
    }

    private void releaseRemoved(String paneId, MemberSession removed, PeerHandle removedHandle,
@@ -30,22 +30,6 @@ class AuthzTest {
        assertTrue(Authz.isUnauthenticated(null));
    }

-    /**
-     * {@code MessageService.answer}'s turn-ownership check treats a caller with no terminal as
-     * matching a turn recorded for the unnamed primary. That rule only stays safe because an
-     * unauthenticated caller — whose terminal is also {@code null} — never reaches {@code answer}
-     * at all: {@link #anonymousIsAuthorizedForNothing} already covers every action including
-     * {@code ANSWER}, but this test names the exact coupling so a future change to either side
-     * cannot drift without turning this test red.
-     */
-    @Test
-    void anAnonymousCallerIsRefusedAnswerSoItCanNeverBeMistakenForTheUnnamedPrimary() {
-        assertFalse(Authz.permits(ANON, ANSWER, null),
-                "an anonymous caller, whose terminal is also null, must never reach answer() — the "
-                        + "turn-ownership check's null-terminal match for the unnamed primary owner "
-                        + "relies on this gate refusing it first");
-    }
-
    @Test
    void orchestrationBelongsToThePrimaryAlone() {
        for (Authz.Action a : new Authz.Action[]{SPAWN, STOP, SEND, DRAIN}) {
@@ -1,13 +1,25 @@
 package dev.ltms.fleet.auth;

+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.mcp.ConnectionIdentity;
 import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
 import dev.ltms.fleet.peer.MemberRole;
+import dev.ltms.fleet.session.MemberSession;
+import dev.ltms.fleet.session.SessionManager;
+import dev.ltms.fleet.session.WorktreeRequest;
+import dev.ltms.fleet.session.Worktrees;
 import org.junit.jupiter.api.Test;

+import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;

 import static org.junit.jupiter.api.Assertions.*;

@@ -541,6 +553,219 @@ class CallerResolverTest {
        assertEquals("term_a", p.terminal());
    }

+    /**
+     * fleetd #702: between {@link SessionManager#release} removing the registry entry and the
+     * pane actually stopping, a resolve for that terminal must still see the live member and
+     * never fall through to a tab map — wired through the real {@link SessionManager}, not a
+     * hand-rolled stand-in for its {@code spawnedMemberRole}.
+     *
+     * <p>Reuses the ticket's own test idea: an injected {@code hasUncommitted} resolves the
+     * releasing terminal from inside {@code release}'s window — a real call landing inside the
+     * window, so no sleep and no race.
+     *
+     * <p>The property asserted is "no tab map is consulted", never "the same role is returned".
+     * The mandatory control is the lead tab map: it names this exact terminal, and the same
+     * resolve taken <em>outside</em> the window (before release runs) must still return the lead
+     * role — without that control, the in-window assertion would also pass on an empty map and
+     * prove nothing.
+     */
+    @Test
+    void aPaneMidTeardownResolvesAsItsOwnRoleConsultingNoTabMap() {
+        FakeHerdr herdr = new FakeHerdr().pinNextStarts(1, "term_a", "w2:p7");
+        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);
+
+        AtomicReference<Principal> duringWindow = new AtomicReference<>();
+        AtomicReference<CallerResolver> resolverRef = new AtomicReference<>();
+        Worktrees worktrees = new Worktrees() {
+            @Override
+            public String add(String repoRoot, String branch, String baseRef) {
+                return "/wt/" + branch.replace('/', '_');
+            }
+
+            @Override
+            public void remove(String repoRoot, String worktreePath) {
+            }
+
+            @Override
+            public void deleteBranch(String repoRoot, String branch) {
+            }
+
+            @Override
+            public boolean hasUncommitted(String worktreePath) {
+                // Runs from INSIDE release()'s git-status shell-out: the registry entry is
+                // already gone, but the pane has not stopped yet.
+                duringWindow.set(resolverRef.get().resolve("127.0.0.1", 42, null));
+                return false;
+            }
+
+            @Override
+            public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+            }
+
+            @Override
+            public String repoRoot(String cwd) {
+                return "/repo";
+            }
+
+            @Override
+            public Optional<String> snapshot(String worktreePath, String branch, String message) {
+                return Optional.empty();
+            }
+
+            @Override
+            public WipRefStats wipRefs(String repoRoot) {
+                return new WipRefStats(0, 0L);
+            }
+
+            @Override
+            public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+                return 0;
+            }
+
+            @Override
+            public void shareWithGroup(String repoRoot, String worktreePath) {
+            }
+        };
+        SessionManager sessions = new SessionManager(workers, worktrees);
+
+        // The lead tab map names "term_a" before anything is ever spawned onto it — the control
+        // this test needs. Built up front so the SAME resolver answers every resolve() call below.
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> FakeHerdr.WORKER_PID);
+        CallerResolver resolver = CallerResolver.withLeadsAndMembers(identity, false, null,
+                () -> Map.of("term_a", "the-lead"), null, sessions::spawnedMemberRole, Map::of);
+        resolverRef.set(resolver);
+
+        Principal before = resolver.resolve("127.0.0.1", 42, null);
+        assertEquals(Role.PRIMARY, before.role(),
+                "control: with no live or releasing member on this terminal, the lead tab map must "
+                        + "win — this is what proves the in-window assertion below is not passing on "
+                        + "an empty map");
+        assertEquals("the-lead", before.name());
+
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702", null));
+        assertEquals("term_a", s.terminalId(), "sanity: the spawn resolved to the pinned pane");
+
+        sessions.release(s.paneId());
+
+        assertNotNull(duringWindow.get(), "the dirty check must have run and captured a resolve");
+        assertEquals(Role.WORKER, duringWindow.get().role(),
+                "inside the window the pane must resolve as its own live-member role, consulting no "
+                        + "tab map — a lead tab naming the same terminal must not win");
+        assertEquals("term_a", duringWindow.get().terminal());
+    }
+
+    /**
+     * {@link SessionManager#releaseRemoved} unbinds the architect slot before the git-status
+     * shell-out that opens the teardown window, so a resolve landing inside that window must see
+     * the slot already unbound and resolve {@link Role#WORKER} — never {@link Role#ARCHITECT},
+     * and never by checking role equality against the live session, which would hold even if the
+     * unbind ran too late.
+     *
+     * <p>The control is the same resolve taken outside the window, while the slot is still bound,
+     * which must return {@link Role#ARCHITECT} — without it this test would also pass against a
+     * slot that was never bound, and prove nothing.
+     */
+    @Test
+    void aReleasingArchitectIsDemotedToWorkerInsideTheTeardownWindow() {
+        FakeHerdr herdr = new FakeHerdr().pinNextStarts(1, "term_a", "w2:p7");
+        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);
+
+        AtomicReference<Principal> duringWindow = new AtomicReference<>();
+        AtomicReference<CallerResolver> resolverRef = new AtomicReference<>();
+        Worktrees worktrees = new Worktrees() {
+            @Override
+            public String add(String repoRoot, String branch, String baseRef) {
+                return "/wt/" + branch.replace('/', '_');
+            }
+
+            @Override
+            public void remove(String repoRoot, String worktreePath) {
+            }
+
+            @Override
+            public void deleteBranch(String repoRoot, String branch) {
+            }
+
+            @Override
+            public boolean hasUncommitted(String worktreePath) {
+                // Runs from INSIDE release()'s git-status shell-out: the architect slot is already
+                // unbound by this point, but the pane has not stopped yet.
+                duringWindow.set(resolverRef.get().resolve("127.0.0.1", 42, null));
+                return false;
+            }
+
+            @Override
+            public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+            }
+
+            @Override
+            public String repoRoot(String cwd) {
+                return "/repo";
+            }
+
+            @Override
+            public Optional<String> snapshot(String worktreePath, String branch, String message) {
+                return Optional.empty();
+            }
+
+            @Override
+            public WipRefStats wipRefs(String repoRoot) {
+                return new WipRefStats(0, 0L);
+            }
+
+            @Override
+            public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+                return 0;
+            }
+
+            @Override
+            public void shareWithGroup(String repoRoot, String worktreePath) {
+            }
+        };
+        SessionManager sessions = new SessionManager(workers, worktrees);
+
+        MemberRegistry members = new MemberRegistry(new FleetConfig.Fleet(Map.of(),
+                Map.of("lead-designer", new FleetConfig.Slot("ltms-local")), Map.of(), Map.of(), null));
+        sessions.setMemberLifecycle(members);
+
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> FakeHerdr.WORKER_PID);
+        CallerResolver resolver = CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, members, sessions::spawnedMemberRole, Map::of);
+        resolverRef.set(resolver);
+
+        MemberSession s = sessions.acquire("ltms-local", MemberRole.ARCHITECT, null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702d", null));
+        assertEquals("term_a", s.terminalId(), "sanity: the spawn resolved to the pinned pane");
+        assertEquals(MemberRole.ARCHITECT, s.role(), "sanity: the slot bind succeeded");
+
+        Principal before = resolver.resolve("127.0.0.1", 42, null);
+        assertEquals(Role.ARCHITECT, before.role(),
+                "control: with the slot still bound, the pane must resolve as an architect — this "
+                        + "is what proves the in-window assertion below is not passing against a "
+                        + "slot that was never bound");
+        assertEquals("lead-designer", before.name());
+
+        sessions.release(s.paneId());
+
+        assertNotNull(duringWindow.get(), "the dirty check must have run and captured a resolve");
+        assertEquals(Role.WORKER, duringWindow.get().role(),
+                "inside the window the architect slot is already unbound, so the result must be "
+                        + "WORKER — asserting role-equality with the live session here would tempt "
+                        + "moving the unbind earlier or later, which would be wrong either way");
+        assertEquals("term_a", duringWindow.get().terminal());
+    }
+
    /** A spawned architect in the roster resolves ARCHITECT, carrying its bound slot's name. */
    @Test
    void aSpawnedArchitectInTheRosterResolvesArchitectWithItsSlotName() {
@@ -77,7 +77,7 @@ class FleetMcpTest {

    private void assertSendRoundTrips(String target, Set<String> profiles) throws Exception {
        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> FleetMcp.send(messages, target, "hi", 4000L, null, profiles, null));
+                () -> FleetMcp.send(messages, target, "hi", 4000L, null, profiles));
        long deadline = System.currentTimeMillis() + 3000;
        while (!rendezvous.isWaiting(target) && System.currentTimeMillis() < deadline) {
            Thread.sleep(5);
@@ -103,7 +103,7 @@ class FleetMcpTest {
    void sendThenReplyRoundTrips() throws Exception {
        // fleet_send blocks; fleet_reply resolves it with the worker's structured answer.
        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> FleetMcp.send(messages, "term_a", "review this", 4000L, null, Set.of(), null));
+                () -> FleetMcp.send(messages, "term_a", "review this", 4000L, null, Set.of()));

        // Wait until the send has opened its waiter so the reply resolves it (CB-307: reply now
        // queues in the inbox if no waiter is open, which would break the round-trip).
@@ -181,7 +181,7 @@ class FleetMcpTest {
        String turnId = afterTurnId.substring(0, afterTurnId.indexOf('"'));

        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
-                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L, null));
+                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L));
        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));

        deadline = System.currentTimeMillis() + 3000;
@@ -279,7 +279,7 @@ class FleetMcpTest {
        assertEquals("late reply", messages.drainReplies("term_a").getFirst().content());

        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
-                () -> FleetMcp.answer(messages, firstTurnId, "config.yaml", 5000L, null));
+                () -> FleetMcp.answer(messages, firstTurnId, "config.yaml", 5000L));
        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
            Thread.sleep(5);
@@ -288,89 +288,6 @@ class FleetMcpTest {
        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
    }

-    // --- fleetd #715: fleet_send{turnId} is gated on the caller that owns the turn -------------
-
-    /**
-     * A blocking {@code fleet_send} from one caller opens the turn; a {@code fleet_send{turnId}}
-     * from a different caller is refused as an error, and the real owner's answer still succeeds.
-     */
-    @Test
-    void aDifferentCallersMcpAnswerIsRefusedForABlockingSendButTheRealOwnerSucceeds() throws Exception {
-        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> FleetMcp.send(messages, T, "do X", 5000L, null, Set.of(), "term_owner"));
-        long deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        assertTrue(rendezvous.isWaiting(T));
-
-        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
-                () -> FleetMcp.ask(messages, T, "which config?", 5000L));
-        McpSchema.CallToolResult question = send.get(5, TimeUnit.SECONDS);
-        assertTrue(textOf(question).contains("[question]"), textOf(question));
-        String questionText = textOf(question);
-        String afterTurnId = questionText.substring(questionText.indexOf("turnId=\"") + "turnId=\"".length());
-        String turnId = afterTurnId.substring(0, afterTurnId.indexOf('"'));
-
-        McpSchema.CallToolResult hijacked = FleetMcp.answer(messages, turnId, "evil.yaml", 500L, "term_attacker");
-        assertTrue(hijacked.isError(), "a caller that did not open this turn must get an error, not an answer");
-        assertFalse(ask.isDone(), "a refused answer must not resolve the worker's blocked fleet_ask");
-
-        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
-                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L, "term_owner"));
-        assertEquals("config.yaml", textOf(ask.get(5, TimeUnit.SECONDS)));
-        deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        FleetMcp.reply(messages, T, Role.WORKER, "done");
-        assertEquals("done", textOf(answer.get(5, TimeUnit.SECONDS)));
-    }
-
-    /**
-     * Same hijack and control through the fire-and-poll ({@code sendAsync}) path: the owner comes
-     * from the ticket's recorded creator terminal, not from a caller threaded through a live call.
-     */
-    @Test
-    void aDifferentCallersMcpAnswerIsRefusedForAnAsyncSendButTheRealOwnerSucceeds() throws Exception {
-        McpSchema.CallToolResult accepted =
-                FleetMcp.sendAsync(messages, T, "do it", null, Set.of(), "term_owner");
-        String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
-
-        long deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        assertTrue(rendezvous.isWaiting(T));
-
-        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
-                () -> FleetMcp.ask(messages, T, "which config?", 5000L));
-        MessageService.TaskView asking = messages.poll(ticket);
-        deadline = System.currentTimeMillis() + 3000;
-        while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-            asking = messages.poll(ticket);
-        }
-        assertEquals(MessageService.Phase.ASKING, asking.phase());
-        String turnId = asking.turnId();
-
-        McpSchema.CallToolResult hijacked = FleetMcp.answer(messages, turnId, "evil.yaml", 500L, "term_attacker");
-        assertTrue(hijacked.isError(), "a caller that did not create this delegation must get an error");
-        assertFalse(ask.isDone(), "a refused answer must not resolve the worker's blocked fleet_ask");
-        assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase(),
-                "a refused answer must not advance the async ticket's phase");
-
-        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
-                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L, "term_owner"));
-        assertEquals("config.yaml", textOf(ask.get(5, TimeUnit.SECONDS)));
-        deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        FleetMcp.reply(messages, T, Role.WORKER, "done");
-        assertEquals("done", textOf(answer.get(5, TimeUnit.SECONDS)));
-    }
-
    @Test
    void pollUnknownTicketIsAnError() {
        McpSchema.CallToolResult res = FleetMcp.poll(messages, "task-999", null);
@@ -380,7 +297,7 @@ class FleetMcpTest {

    @Test
    void sendTimesOutWithAWorkingNote() {
-        McpSchema.CallToolResult res = FleetMcp.send(messages, "term_a", "hi", 120L, null, Set.of(), null);
+        McpSchema.CallToolResult res = FleetMcp.send(messages, "term_a", "hi", 120L, null, Set.of());
        assertNotEquals(Boolean.TRUE, res.isError(), "a timeout is informational, not a tool error");
        assertTrue(textOf(res).contains("no reply"), "got: " + textOf(res));
    }
@@ -396,7 +313,7 @@ class FleetMcpTest {
    void sendTimesOutWithAnUnconfirmedNoteNotARetryInvitation() throws Exception {
        herdr.agentSendFailsWith("send_failed");
        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> FleetMcp.send(messages, T, "hi", 150L, null, Set.of(), null));
+                () -> FleetMcp.send(messages, T, "hi", 150L, null, Set.of()));
        long deadline = System.currentTimeMillis() + 2000;
        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
            //noinspection BusyWait
@@ -416,13 +333,13 @@ class FleetMcpTest {

    @Test
    void sendRejectsMissingArgs() {
-        assertTrue(FleetMcp.send(messages, null, "hi", null, null, Set.of(), null).isError());
-        assertTrue(FleetMcp.send(messages, "term_a", "  ", null, null, Set.of(), null).isError());
+        assertTrue(FleetMcp.send(messages, null, "hi", null, null, Set.of()).isError());
+        assertTrue(FleetMcp.send(messages, "term_a", "  ", null, null, Set.of()).isError());
    }

    @Test
    void sendRejectsAConfiguredProfileNameBeforeAcceptingIt() {
-        McpSchema.CallToolResult blocking = FleetMcp.send(messages, "sol", "hi", 100L, null, Set.of("sol"), null);
+        McpSchema.CallToolResult blocking = FleetMcp.send(messages, "sol", "hi", 100L, null, Set.of("sol"));
        McpSchema.CallToolResult async = FleetMcp.sendAsync(messages, "sol", "hi", null, Set.of("sol"));

        assertTrue(blocking.isError());
@@ -441,7 +358,7 @@ class FleetMcpTest {
        assertSendRoundTrips("term_live_member", profiles);

        // A herdr-owned pane outside the bridge roster cannot be classified at accept time.
-        McpSchema.CallToolResult result = FleetMcp.send(messages, "external-pane", "hi", 10L, null, profiles, null);
+        McpSchema.CallToolResult result = FleetMcp.send(messages, "external-pane", "hi", 10L, null, profiles);
        assertFalse(result.isError(), "an unclassified target must not be rejected at acceptance time");
    }

@@ -533,7 +450,7 @@ class FleetMcpTest {
    void askThenAnswerRoundTrips() throws Exception {
        // The primary delegates and blocks; wait until its waiter is open before the worker asks.
        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> FleetMcp.send(messages, "term_a", "do X", 5000L, null, Set.of(), null));
+                () -> FleetMcp.send(messages, "term_a", "do X", 5000L, null, Set.of()));
        long deadline = System.currentTimeMillis() + 3000;
        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
            //noinspection BusyWait
@@ -555,7 +472,7 @@ class FleetMcpTest {

        // The primary answers via fleet_send(turnId); this blocks again for the worker's reply.
        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
-                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L, null));
+                () -> FleetMcp.answer(messages, turnId, "config.yaml", 5000L));

        // The worker's ask returns the answer — it resumes the same turn.
        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
@@ -581,7 +498,7 @@ class FleetMcpTest {

    @Test
    void answerToAStaleTurnIsAnError() {
-        McpSchema.CallToolResult res = FleetMcp.answer(messages, "term_a#999", "too late", 500L, null);
+        McpSchema.CallToolResult res = FleetMcp.answer(messages, "term_a#999", "too late", 500L);
        assertTrue(res.isError());
        assertTrue(textOf(res).contains("no longer open"), textOf(res));
    }
@@ -1984,7 +1901,7 @@ class FleetMcpTest {
        // Clean up the still-open ask so the background thread does not linger past the test.
        String turnId = asking.turnId();
        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(turnId, "config.yaml", 5000, null));
+                () -> messages.answer(turnId, "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        deadline = System.currentTimeMillis() + 3000;
        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
@@ -2041,7 +1958,7 @@ class FleetMcpTest {
        // Clean up the still-open ask so the background thread does not linger past the test.
        String turnId = asking.turnId();
        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(turnId, "config.yaml", 5000, "term_creator"));
+                () -> messages.answer(turnId, "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        deadline = System.currentTimeMillis() + 3000;
        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
@@ -78,7 +78,7 @@ class MessageServiceTest {
    }

    private CompletableFuture<MessageService.Reply> sendAsync(String content, long timeoutMillis) {
-        return CompletableFuture.supplyAsync(() -> messages.send(T, content, timeoutMillis, null));
+        return CompletableFuture.supplyAsync(() -> messages.send(T, content, timeoutMillis));
    }

    private void awaitWaiting() throws InterruptedException {
@@ -188,7 +188,7 @@ class MessageServiceTest {
                localRendezvous, new InMemoryReplyInbox());
        String brief = "Implement the requested change. ".repeat(20);
        CompletableFuture<MessageService.Reply> send =
-                CompletableFuture.supplyAsync(() -> localMessages.send(T, brief, 5000, (String) null));
+                CompletableFuture.supplyAsync(() -> localMessages.send(T, brief, 5000));
        long deadline = System.currentTimeMillis() + 2000;
        while (!localRendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
            Thread.sleep(5);
@@ -341,7 +341,7 @@ class MessageServiceTest {

        // The primary answers via fleet_send(turnId); this blocks again for the worker's reply.
        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));

        // The worker's ask returns the answer — it resumes the same turn.
        MessageService.AskResult a = ask.get(5, TimeUnit.SECONDS);
@@ -377,7 +377,7 @@ class MessageServiceTest {

        // The primary answers that one turnId; both asks unblock with the same answer.
        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));

        MessageService.AskResult a1 = ask1.get(5, TimeUnit.SECONDS);
        MessageService.AskResult a2 = ask2.get(5, TimeUnit.SECONDS);
@@ -472,7 +472,7 @@ class MessageServiceTest {
                "the duplicate's own timeout elapses first");

        CompletableFuture<MessageService.Reply> answered =
-                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "fleetd.yaml", 500, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "fleetd.yaml", 500));

        MessageService.AskResult a = fresh.get(5, TimeUnit.SECONDS);
        assertEquals(MessageService.AskOutcome.ANSWERED, a.outcome(),
@@ -507,7 +507,7 @@ class MessageServiceTest {

        CompletableFuture<MessageService.Reply> lateAnswer = new CompletableFuture<>();
        messages.setAskTimeoutRaceHookForTest(() ->
-                lateAnswer.complete(messages.answer(turnId, "too late", 500, null)));
+                lateAnswer.complete(messages.answer(turnId, "too late", 500)));
        try {
            MessageService.AskResult a = ask.get(5, TimeUnit.SECONDS);
            assertEquals(MessageService.AskOutcome.TIMED_OUT, a.outcome());
@@ -539,7 +539,7 @@ class MessageServiceTest {

    @Test
    void answeringAnUnknownTurnIsStale() {
-        MessageService.Reply r = messages.answer(T + "#999", "too late", 500, null);
+        MessageService.Reply r = messages.answer(T + "#999", "too late", 500);
        assertEquals(MessageService.Outcome.STALE_TURN, r.outcome(),
                "an answer to a turn that never existed (or already lapsed) is stale, not a hang");
    }
@@ -572,7 +572,7 @@ class MessageServiceTest {

        messages.setAnswerAskLapseRaceHookForTest(() -> rendezvous.answerAsk(turnId, "raced in first"));
        try {
-            MessageService.Reply r = messages.answer(turnId, "too late", 500, null);
+            MessageService.Reply r = messages.answer(turnId, "too late", 500);
            assertEquals(MessageService.Outcome.STALE_TURN, r.outcome(),
                    "an ask already answered by the race must be seen as lapsed, not double-delivered");
        } finally {
@@ -596,7 +596,7 @@ class MessageServiceTest {
    void sendTimesOutBeforeDeliveryIsQueuedNotWorking() {
        // Nothing ever delivers the message and nothing resolves the send, so the reply future
        // times out with delivery still incomplete — the message is still queued for the worker.
-        MessageService.Reply r = messages.send(T, "never delivered", 50, null);
+        MessageService.Reply r = messages.send(T, "never delivered", 50);
        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, r.outcome(),
                "an undelivered send that times out is still queued, not working");
        assertNull(r.text());
@@ -605,7 +605,7 @@ class MessageServiceTest {
    @Test
    void sendTimesOutAfterDeliveryIsStillWorking() throws Exception {
        CompletableFuture<MessageService.Reply> send =
-                CompletableFuture.supplyAsync(() -> messages.send(T, "do the task", 300, null));
+                CompletableFuture.supplyAsync(() -> messages.send(T, "do the task", 300));
        awaitWaiting();
        injector.onStatus(T, AgentStatus.IDLE);    // deliver — the delivered future now completes
        injector.onStatus(T, AgentStatus.WORKING); // worker starts but never replies
@@ -630,7 +630,7 @@ class MessageServiceTest {
    void sendTimeoutUsesCancellationDeliveredWhenPickupWinsTheRace() {
        messages.setTimeoutCancellationRaceHookForTest(() -> injector.onStatus(T, AgentStatus.IDLE));
        try {
-            MessageService.Reply reply = messages.send(T, "race delivery", 50, null);
+            MessageService.Reply reply = messages.send(T, "race delivery", 50);

            assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, reply.outcome(),
                    "cancel reporting DELIVERED means the worker received the timed-out message");
@@ -652,7 +652,7 @@ class MessageServiceTest {
    void sendTimesOutWithAttemptedDeliveryReportsUnconfirmedNotQueued() throws Exception {
        herdr.agentSendFailsWith("send_failed");
        CompletableFuture<MessageService.Reply> send =
-                CompletableFuture.supplyAsync(() -> messages.send(T, "brief", 150, null));
+                CompletableFuture.supplyAsync(() -> messages.send(T, "brief", 150));
        awaitWaiting();
        injector.onStatus(T, AgentStatus.IDLE); // triggers the failing delivery attempt → ATTEMPTED

@@ -679,7 +679,7 @@ class MessageServiceTest {

        // The primary answers, unblocking the worker; but the worker never sends the follow-up
        // fleet_reply, so the answering send rides out its short window as still-working.
-        MessageService.Reply answer = messages.answer(q.turnId(), "config.yaml", 200, null);
+        MessageService.Reply answer = messages.answer(q.turnId(), "config.yaml", 200);
        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answer.outcome(),
                "an answered worker that never replies times out as still working");

@@ -715,7 +715,7 @@ class MessageServiceTest {
        assertEquals(MessageService.Outcome.QUESTION, q.outcome());

        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));
        ask.get(5, TimeUnit.SECONDS); // worker resumed with the answer

        awaitWaiting(); // the answering call has (re)opened its own forward waiter
@@ -723,7 +723,7 @@ class MessageServiceTest {
        MessageService.Reply done = answer.get(5, TimeUnit.SECONDS);
        assertEquals(MessageService.Outcome.REPLIED, done.outcome());

-        MessageService.Reply probe = messages.send(T, "probe after normal reply", 300, null);
+        MessageService.Reply probe = messages.send(T, "probe after normal reply", 300);
        assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
                "the session lock must be released after a normal REPLIED answer(), or this bounded "
                        + "follow-up send would come back BUSY instead of timing out on its own work");
@@ -756,13 +756,13 @@ class MessageServiceTest {
        // The primary answers, unblocking the worker; the worker never sends its follow-up
        // fleet_reply, so the answering call rides out its short window as still-working.
        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 200, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 200));
        MessageService.Reply answered = answer.get(5, TimeUnit.SECONDS);
        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answered.outcome(),
                "an answered worker that never replies times out as still working");
        ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer

-        MessageService.Reply probe = messages.send(T, "probe after timed-out-working", 300, null);
+        MessageService.Reply probe = messages.send(T, "probe after timed-out-working", 300);
        assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
                "the session lock must be released after a TIMED_OUT_WORKING answer(), or this bounded "
                        + "follow-up send would come back BUSY instead of timing out on its own work");
@@ -791,7 +791,7 @@ class MessageServiceTest {
                new java.util.concurrent.atomic.AtomicReference<>();
        Thread answerer = new Thread(() -> {
            try {
-                messages.answer(turnId, "config.yaml", 5000, null);
+                messages.answer(turnId, "config.yaml", 5000);
                caught.set(new AssertionError("expected answer() to throw"));
            } catch (Throwable t) {
                caught.set(t);
@@ -813,7 +813,7 @@ class MessageServiceTest {

        ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer

-        MessageService.Reply probe = messages.send(T, "probe after exceptional failure", 300, null);
+        MessageService.Reply probe = messages.send(T, "probe after exceptional failure", 300);
        assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
                "the session lock must be released when answer()'s reply future fails exceptionally, "
                        + "or this bounded follow-up send would come back BUSY instead of timing out on its own work");
@@ -840,7 +840,7 @@ class MessageServiceTest {
                new java.util.concurrent.atomic.AtomicReference<>();
        Thread answerer = new Thread(() -> {
            try {
-                messages.answer(turnId, "config.yaml", 5000, null);
+                messages.answer(turnId, "config.yaml", 5000);
                caught.set(new AssertionError("expected answer() to throw"));
            } catch (Throwable t) {
                caught.set(t);
@@ -859,7 +859,7 @@ class MessageServiceTest {

        ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer

-        MessageService.Reply probe = messages.send(T, "probe after interruption", 300, null);
+        MessageService.Reply probe = messages.send(T, "probe after interruption", 300);
        assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
                "the session lock must be released when answer()'s wait is interrupted, or this bounded "
                        + "follow-up send would come back BUSY instead of timing out on its own work");
@@ -967,11 +967,11 @@ class MessageServiceTest {
    @Test
    void concurrentSendToSameSessionWhileFirstHoldsItIsBusy() throws Exception {
        CompletableFuture<MessageService.Reply> first =
-                CompletableFuture.supplyAsync(() -> messages.send(T, "first", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.send(T, "first", 5000));
        awaitWaiting(); // the first send now holds the session lock, blocked on its reply

        // A second send to the SAME session cannot take the lock within its short window.
-        MessageService.Reply busy = messages.send(T, "second", 100, null);
+        MessageService.Reply busy = messages.send(T, "second", 100);
        assertEquals(MessageService.Outcome.BUSY, busy.outcome(),
                "a second send while another holds the session is busy, not a hang");
        assertNull(busy.text());
@@ -1001,13 +1001,13 @@ class MessageServiceTest {
        PrimaryRegistry reg = new PrimaryRegistry(null);
        // L accepts a delegation to W: the send wins the lock and queues delivery → L is recorded.
        CompletableFuture<MessageService.Reply> first = CompletableFuture.supplyAsync(
-                () -> messages.send(T, "first", 5000, () -> reg.recordDelegation(T, LEAD_L), null));
+                () -> messages.send(T, "first", 5000, () -> reg.recordDelegation(T, LEAD_L)));
        awaitWaiting();
        assertEquals(LEAD_L, reg.nudgeTargetFor(T).orElseThrow(),
                "an accepted send owns the delegation");

        // A attempts W while L holds it → BUSY (lock never taken) → its hook never fires.
-        MessageService.Reply busy = messages.send(T, "second", 100, () -> reg.recordDelegation(T, LEAD_A), null);
+        MessageService.Reply busy = messages.send(T, "second", 100, () -> reg.recordDelegation(T, LEAD_A));
        assertEquals(MessageService.Outcome.BUSY, busy.outcome());
        assertEquals(LEAD_L, reg.nudgeTargetFor(T).orElseThrow(),
                "a BUSY send must not steal the delegator ownership it never earned");
@@ -1028,7 +1028,7 @@ class MessageServiceTest {
    void anAcceptedSendAfterThePriorOwnerFinishesBecomesTheNewOwner() throws Exception {
        PrimaryRegistry reg = new PrimaryRegistry(null);
        CompletableFuture<MessageService.Reply> first = CompletableFuture.supplyAsync(
-                () -> messages.send(T, "first", 5000, () -> reg.recordDelegation(T, LEAD_L), null));
+                () -> messages.send(T, "first", 5000, () -> reg.recordDelegation(T, LEAD_L)));
        awaitWaiting();
        injector.onStatus(T, AgentStatus.IDLE);
        injector.onStatus(T, AgentStatus.WORKING);
@@ -1039,7 +1039,7 @@ class MessageServiceTest {

        // L finished; A's later accepted send takes the delegation over.
        CompletableFuture<MessageService.Reply> second = CompletableFuture.supplyAsync(
-                () -> messages.send(T, "second", 5000, () -> reg.recordDelegation(T, LEAD_A), null));
+                () -> messages.send(T, "second", 5000, () -> reg.recordDelegation(T, LEAD_A)));
        awaitWaiting();
        assertEquals(LEAD_A, reg.nudgeTargetFor(T).orElseThrow(),
                "an accepted send after the owner finished becomes the new delegator");
@@ -1060,7 +1060,7 @@ class MessageServiceTest {
    void answeringAnAskDoesNotRewriteDelegatorOwnership() throws Exception {
        PrimaryRegistry reg = new PrimaryRegistry(null);
        CompletableFuture<MessageService.Reply> send = CompletableFuture.supplyAsync(
-                () -> messages.send(T, "do X", 5000, () -> reg.recordDelegation(T, LEAD_L), null));
+                () -> messages.send(T, "do X", 5000, () -> reg.recordDelegation(T, LEAD_L)));
        awaitWaiting();
        assertEquals(LEAD_L, reg.nudgeTargetFor(T).orElseThrow(), "L owns the delegation");

@@ -1075,7 +1075,7 @@ class MessageServiceTest {

        // L answers the ask on the same turn; the answer path must not touch ownership.
        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting(); // the answering send reopened its forward waiter
        assertEquals(LEAD_L, reg.nudgeTargetFor(T).orElseThrow(),
@@ -1095,7 +1095,7 @@ class MessageServiceTest {
    void aThrowingAcceptedHookLeavesNoStaleWaiterOrQueuedOrphan() {
        assertThrows(IllegalStateException.class,
                () -> messages.send(T, "doomed", 500,
-                        () -> { throw new IllegalStateException("ownership hook failed"); }, null),
+                        () -> { throw new IllegalStateException("ownership hook failed"); }),
                "a throwing ownership hook fails the send loudly");

        assertFalse(rendezvous.isWaiting(T), "the failed send must not leave a stale rendezvous waiter");
@@ -1223,7 +1223,7 @@ class MessageServiceTest {
    @Test
    void abandonFailsASendThatIsStillWaitingOnAReleasedSession() throws Exception {
        CompletableFuture<MessageService.Reply> send =
-                CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000, null));
+                CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000));
        awaitWaiting();

        assertTrue(messages.abandon(T, "session released"), "a live waiter is abandoned");
@@ -1243,7 +1243,7 @@ class MessageServiceTest {
    @Test
    void abandonDoesNotOverwriteAnAlreadyResolvedSend() throws Exception {
        CompletableFuture<MessageService.Reply> send =
-                CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000, null));
+                CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000));
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "the real answer"));

@@ -1374,7 +1374,7 @@ class MessageServiceTest {
        assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase());

        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, null));
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "done"));
@@ -1414,7 +1414,7 @@ class MessageServiceTest {
        // same turnId must see it as lapsed rather than resolving a question nobody is waiting on.
        assertEquals(MessageService.AskOutcome.TIMED_OUT, ask.get(5, TimeUnit.SECONDS).outcome());
        assertEquals(MessageService.Outcome.STALE_TURN,
-                messages.answer(asking.turnId(), "config.yaml", 200, null).outcome());
+                messages.answer(asking.turnId(), "config.yaml", 200).outcome());
    }

    @Test
@@ -1451,7 +1451,7 @@ class MessageServiceTest {

        // The primary answers, but its own bounded wait for the worker's resumed turn is short and
        // expires before the worker (still genuinely working) gets back to it.
-        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150, null);
+        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150);
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
                "the primary's own bounded wait gives up before the worker finishes resuming");
@@ -1480,7 +1480,7 @@ class MessageServiceTest {
                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);

-        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150, null);
+        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150);
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome());

@@ -1534,7 +1534,7 @@ class MessageServiceTest {
        messages.setFinishAsyncTaskRaceHookForTest(() -> messages.forgetTurnForTest(turnId));

        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting(); // answer() opened its own forward waiter for the resumed worker turn

@@ -1640,7 +1640,7 @@ class MessageServiceTest {
        String turnId = asking.turnId();

        CompletableFuture<MessageService.Reply> answer =
-                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000, null));
+                CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer(),
                "the worker's own ask() call must have already unblocked with the primary's answer "
                        + "before we force the race below");
@@ -1687,7 +1687,7 @@ class MessageServiceTest {
        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
        String turnId = asking.turnId();

-        MessageService.Reply answerReply = messages.answer(turnId, "config.yaml", 150, null);
+        MessageService.Reply answerReply = messages.answer(turnId, "config.yaml", 150);
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
                "the primary's own bounded wait must give up first, leaving turnId stamped with no "
@@ -1753,7 +1753,7 @@ class MessageServiceTest {

        MessageService.TaskView asking = messages.poll(first);
        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, null));
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "done"));
@@ -1787,7 +1787,7 @@ class MessageServiceTest {

        // The primary answers it — answer() resumes the turn and blocks for what comes next.
        CompletableFuture<MessageService.Reply> answer1 = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking1.turnId(), "a1", 5000, null));
+                () -> messages.answer(asking1.turnId(), "a1", 5000));
        assertEquals("a1", ask1.get(5, TimeUnit.SECONDS).answer());

        // Still in the SAME resumed turn — before replying — the worker asks again.
@@ -1808,7 +1808,7 @@ class MessageServiceTest {

        // The primary answers the second question; the worker finally sends its real fleet_reply.
        CompletableFuture<MessageService.Reply> answer2 = CompletableFuture.supplyAsync(
-                () -> messages.answer(turnId2, "a2", 5000, null));
+                () -> messages.answer(turnId2, "a2", 5000));
        assertEquals("a2", ask2.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "done"));
@@ -1908,7 +1908,7 @@ class MessageServiceTest {
        assertEquals(asking.turnId(), pending.turnId());

        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, null));
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        assertNull(messages.pendingAsk(T, null), "an answered question is no longer pending");
        awaitWaiting();
@@ -1943,7 +1943,7 @@ class MessageServiceTest {
        assertEquals("which config file?", unnamed.question());

        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, "term_creator"));
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "done"));
@@ -1971,81 +1971,13 @@ class MessageServiceTest {
                "the unnamed primary must still see it even with no recorded creator");

        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, null));
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
        awaitWaiting();
        assertTrue(rendezvous.resolve(T, "done"));
        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
    }

-    // --- fleetd #715: answer() is gated on the caller that owns the turn -----------------------
-
-    /**
-     * A blocking {@code fleet_send} from one caller opens the turn; a different caller's answer is
-     * refused with no side effect on the rendezvous or the worker's blocked {@code fleet_ask} —
-     * only the real owner can answer it.
-     */
-    @Test
-    void aDifferentCallersAnswerIsRefusedForABlockingSendDelegationButTheRealOwnerSucceeds() throws Exception {
-        CompletableFuture<MessageService.Reply> send = CompletableFuture.supplyAsync(
-                () -> messages.send(T, "do X", 5000, "term_owner"));
-        awaitWaiting();
-        injectDelivery();
-
-        CompletableFuture<MessageService.AskResult> ask =
-                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
-        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
-        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
-        assertFalse(rendezvous.isWaiting(T), "the forward waiter closes once the question surfaces");
-
-        // The hijack: a different caller answers the SAME turnId.
-        MessageService.Reply hijacked = messages.answer(q.turnId(), "evil.yaml", 500, "term_attacker");
-        assertEquals(MessageService.Outcome.NOT_TURN_OWNER, hijacked.outcome(),
-                "a caller that did not open this turn must be refused, not served");
-        assertFalse(rendezvous.isWaiting(T), "a refused answer must not open a forward waiter");
-        assertFalse(ask.isDone(), "a refused answer must not resolve the worker's blocked fleet_ask");
-        assertEquals(T, rendezvous.askSession(q.turnId()), "a refused answer must leave the ask turn open");
-
-        // The control: the real owner answers the same turnId and the turn resumes normally.
-        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(q.turnId(), "config.yaml", 5000, "term_owner"));
-        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-        awaitWaiting();
-        assertTrue(rendezvous.resolve(T, "done"));
-        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
-    }
-
-    /**
-     * Same hijack and control as the blocking case, but the delegation is opened through the
-     * fire-and-poll path ({@code sendAsync}) — the owner comes from the ticket's recorded creator
-     * terminal, not a caller argument threaded through a live blocking call.
-     */
-    @Test
-    void aDifferentCallersAnswerIsRefusedForAnAsyncSendDelegationButTheRealOwnerSucceeds() throws Exception {
-        String ticket = messages.sendAsync(T, "task that asks", null, "term_owner");
-        awaitWaiting();
-        injectDelivery();
-
-        CompletableFuture<MessageService.AskResult> ask =
-                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
-        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
-
-        MessageService.Reply hijacked = messages.answer(asking.turnId(), "evil.yaml", 500, "term_attacker");
-        assertEquals(MessageService.Outcome.NOT_TURN_OWNER, hijacked.outcome(),
-                "a caller that did not create this delegation must be refused, not served");
-        assertFalse(ask.isDone(), "a refused answer must not resolve the worker's blocked fleet_ask");
-        assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase(),
-                "a refused answer must not advance the async ticket's phase");
-
-        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000, "term_owner"));
-        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-        awaitWaiting();
-        assertTrue(rendezvous.resolve(T, "done"));
-        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
-        assertEquals("done", awaitTicketPhase(ticket, MessageService.Phase.DONE).reply());
-    }
-
    // --- CB-588: async ticket terminal nudges ---------------------------------------------------
    //
    // MessageService.reply's rendezvous fast path is exactly what an async ticket always takes
@@ -2302,7 +2234,7 @@ class MessageServiceTest {

            // Clean up the still-open ask so the background thread does not linger past the test.
            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000, null));
+                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000));
            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
            awaitWaiting();
            assertTrue(rendezvous.resolve(T, "done"));
@@ -2328,7 +2260,7 @@ class MessageServiceTest {
                    .filter(c -> c.method().equals("agent.prompt")).count();

            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000, null));
+                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000));
            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
            awaitWaiting();
            java.util.concurrent.CountDownLatch terminalReached = new java.util.concurrent.CountDownLatch(1);
@@ -2721,7 +2653,7 @@ class MessageServiceTest {
    void hasQueuedDeliveryIsTrueAfterAnUndeliveredSendTimesOut() {
        // Nothing ever delivers the message (never goes IDLE/BLOCKED), so the send times out with
        // TIMED_OUT_QUEUED — same setup as sendTimesOutBeforeDeliveryIsQueuedNotWorking above.
-        MessageService.Reply r = messages.send(T, "never delivered", 50, null);
+        MessageService.Reply r = messages.send(T, "never delivered", 50);
        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, r.outcome());

        assertTrue(messages.hasQueuedDelivery(T),
@@ -2733,7 +2665,7 @@ class MessageServiceTest {

    @Test
    void hasQueuedDeliveryClearsOnceTheTargetsNextDeliveryIsAccepted() throws Exception {
-        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, messages.send(T, "first", 50, null).outcome());
+        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, messages.send(T, "first", 50).outcome());
        assertTrue(messages.hasQueuedDelivery(T));

        // A fresh send accepts delivery (opens its own waiter) — the stale queued fact is cleared.
@@ -2750,7 +2682,7 @@ class MessageServiceTest {

    @Test
    void hasQueuedDeliveryClearsOnAbandon() {
-        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, messages.send(T, "first", 50, null).outcome());
+        assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, messages.send(T, "first", 50).outcome());
        assertTrue(messages.hasQueuedDelivery(T));

        messages.abandon(T, "session released");
@@ -1,187 +0,0 @@
-package dev.ltms.fleet.msg;
-
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.stream.Stream;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Pins that no file under {@code src/main/java} calls the fail-open
- * {@link Rendezvous#open(String)} overload. That overload opens a forward waiter with no
- * recorded {@link Rendezvous.Owner}, so the turn it opens can never be answered by anyone,
- * not even the unnamed primary. Every production caller must go through
- * {@link Rendezvous#open(String, Rendezvous.Owner)} and record an explicit owner.
- *
- * <p>This reads each file's own source text rather than reflecting on compiled bytecode, because
- * the risk is a future one-word edit at a call site, not a missing overload.
- *
- * <p>The scan below finds a violation by its receiver, {@code rendezvous.open(}, classified by
- * argument count. The one test method here points that exact scanner at a file known to hold
- * many real one-argument calls before it ever looks at production, so a scanner that stops
- * matching fails loudly on the known-positive case instead of leaving a clean production result
- * looking like evidence it never produced.
- */
-class RendezvousOpenUsageTest {
-
-    private static final Path PRODUCTION_SOURCE = Path.of("src/main/java");
-    private static final Path KNOWN_TEST_CALLER =
-            Path.of("src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java");
-
-    /**
-     * A pattern that cannot find a known one-argument {@code rendezvous.open(} call would also
-     * find none in production -- not because production is clean, but because the pattern does
-     * not match the text it is supposed to catch. That failure mode is exactly what let a
-     * {@code \b}-based regex read "no callers" under {@code git grep -E} when 81 real ones
-     * existed: {@code git grep} does not treat {@code \b} as a word boundary, so the pattern
-     * silently matched nothing anywhere, clean code and real calls alike. This test runs the
-     * known-positive check first, with the same scanning method the production check then
-     * depends on, so that mistake fails loudly here instead of reading as a clean result.
-     */
-    @Test
-    void noProductionFileCallsTheSingleArgumentOpenOverload() throws IOException {
-        List<String> knownCalls = new ArrayList<>();
-        int knownFilesScanned = scanForOneArgOpenCalls(KNOWN_TEST_CALLER, knownCalls);
-
-        // CONTROL: the scan actually walked files -- a wrong root would otherwise report "found
-        // nothing" having looked at nothing.
-        assertTrue(knownFilesScanned > 0, "control failed: the scan under " + KNOWN_TEST_CALLER
-                + " visited zero .java files -- the path is wrong, so neither result below proves "
-                + "anything");
-
-        // CONTROL: the scanner actually finds real one-argument rendezvous.open( calls when
-        // pointed at a file known to hold many. If this is not satisfied, the matching logic
-        // itself is broken, and the production result below is the scanner failing silently,
-        // not production code actually being clean.
-        assertTrue(knownCalls.size() >= 60, "control failed: the scanner found only "
-                + knownCalls.size() + " one-argument rendezvous.open( call(s) in " + KNOWN_TEST_CALLER
-                + ", which is known to hold many -- the matching logic itself is broken: " + knownCalls);
-
-        List<String> violations = new ArrayList<>();
-        int filesScanned = scanForOneArgOpenCalls(PRODUCTION_SOURCE, violations);
-
-        // CONTROL: the scan actually walked files -- a wrong root would otherwise report "no
-        // violations found" having looked at nothing.
-        assertTrue(filesScanned > 0, "control failed: the scan under " + PRODUCTION_SOURCE
-                + " visited zero .java files -- the path is wrong, so the absence of violations "
-                + "below proves nothing");
-
-        assertTrue(violations.isEmpty(), "found a call to the fail-open Rendezvous.open(String) "
-                + "overload, which opens a forward waiter with no recorded owner -- record an "
-                + "explicit Rendezvous.Owner through open(String, Owner) instead: " + violations);
-    }
-
-    private static int scanForOneArgOpenCalls(Path root, List<String> sites) throws IOException {
-        List<Path> files = javaFiles(root);
-        for (Path file : files) {
-            scanFileForOpenCalls(file, sites);
-        }
-        return files.size();
-    }
-
-    private static List<Path> javaFiles(Path root) throws IOException {
-        try (Stream<Path> paths = Files.walk(root)) {
-            return paths.filter(p -> p.toString().endsWith(".java")).toList();
-        }
-    }
-
-    private static void scanFileForOpenCalls(Path file, List<String> sites) throws IOException {
-        String source = Files.readString(file);
-        String needle = "rendezvous.open(";
-        int from = 0;
-        int idx;
-        while ((idx = source.indexOf(needle, from)) >= 0) {
-            int argsStart = idx + needle.length();
-            String args = extractBalancedArgs(source, argsStart, file, idx);
-            int closeParenIndex = argsStart + args.length();
-            from = closeParenIndex + 1;
-
-            if (args.isBlank()) {
-                continue; // Rendezvous has no zero-argument open() -- a javadoc "rendezvous.open()" mention, not a call
-            }
-            if (topLevelCommaCount(args) == 0) {
-                sites.add(file + ":" + lineOf(source, idx) + " -- rendezvous.open(" + args.trim() + ")");
-            }
-        }
-    }
-
-    /**
-     * The text between {@code rendezvous.open(} and its matching close paren: balanced over
-     * nested calls, and never split by a paren or comma sitting inside a string or char literal.
-     */
-    private static String extractBalancedArgs(String source, int start, Path file, int callIndex) {
-        int depth = 1;
-        boolean inString = false;
-        boolean inChar = false;
-        int i = start;
-        while (i < source.length()) {
-            char c = source.charAt(i);
-            if (inString) {
-                if (c == '\\') { i += 2; continue; }
-                if (c == '"') inString = false;
-            } else if (inChar) {
-                if (c == '\\') { i += 2; continue; }
-                if (c == '\'') inChar = false;
-            } else if (c == '"') {
-                inString = true;
-            } else if (c == '\'') {
-                inChar = true;
-            } else if (c == '(') {
-                depth++;
-            } else if (c == ')') {
-                depth--;
-                if (depth == 0) return source.substring(start, i);
-            }
-            i++;
-        }
-        throw new IllegalStateException(
-                "unbalanced parentheses scanning " + file + ":" + lineOf(source, callIndex));
-    }
-
-    /**
-     * Commas at paren/bracket/brace depth zero, skipping string and char literals -- the argument
-     * separators a human reader would see, not every comma character in the text.
-     */
-    private static int topLevelCommaCount(String args) {
-        int depth = 0;
-        int commas = 0;
-        boolean inString = false;
-        boolean inChar = false;
-        int i = 0;
-        while (i < args.length()) {
-            char c = args.charAt(i);
-            if (inString) {
-                if (c == '\\') { i += 2; continue; }
-                if (c == '"') inString = false;
-            } else if (inChar) {
-                if (c == '\\') { i += 2; continue; }
-                if (c == '\'') inChar = false;
-            } else if (c == '"') {
-                inString = true;
-            } else if (c == '\'') {
-                inChar = true;
-            } else if (c == '(' || c == '[' || c == '{') {
-                depth++;
-            } else if (c == ')' || c == ']' || c == '}') {
-                depth--;
-            } else if (c == ',' && depth == 0) {
-                commas++;
-            }
-            i++;
-        }
-        return commas;
-    }
-
-    private static int lineOf(String source, int index) {
-        int line = 1;
-        for (int i = 0; i < index; i++) {
-            if (source.charAt(i) == '\n') line++;
-        }
-        return line;
-    }
-}
@@ -132,80 +132,4 @@ class RendezvousTest {
        assertNull(rendezvous.askSession(t.turnId()), "a closed ask is forgotten");
        assertFalse(rendezvous.answerAsk(t.turnId(), "late"), "a closed ask can no longer be answered");
    }
-
-    // ── fleetd #715: turn ownership ────────────────────────────────────────────────────────────
-
-    @Test
-    void openWithNoOwnerRecordsNoOwnerAndOpenWithAnOwnerRecordsIt() {
-        rendezvous.open(W);
-        assertNull(rendezvous.ownerOf(W), "the no-owner overload records no owner at all");
-        rendezvous.close(W, rendezvous.currentWaiter(W));
-
-        rendezvous.open(W, Rendezvous.Owner.of("term_lead"));
-        assertEquals(Rendezvous.Owner.of("term_lead"), rendezvous.ownerOf(W));
-    }
-
-    @Test
-    void ownerOfIsNullWhenNoWaiterIsOpen() {
-        assertNull(rendezvous.ownerOf(W), "no waiter open means no owner to report");
-    }
-
-    @Test
-    void openAskStampsTheForwardWaitersOwnerOntoTheFreshTurnOnly() {
-        rendezvous.open(W, Rendezvous.Owner.of("term_lead"));
-
-        Rendezvous.AskTicket fresh = rendezvous.openAsk(W);
-        assertTrue(fresh.fresh());
-        assertEquals(Rendezvous.Owner.of("term_lead"), rendezvous.askOwner(fresh.turnId()),
-                "a freshly-opened ask copies the forward waiter's current owner");
-
-        Rendezvous.AskTicket coalesced = rendezvous.openAsk(W);
-        assertFalse(coalesced.fresh());
-        assertEquals(fresh.turnId(), coalesced.turnId());
-        assertEquals(Rendezvous.Owner.of("term_lead"), rendezvous.askOwner(coalesced.turnId()),
-                "a coalesced duplicate ask rides the fresh owner's turn, unchanged");
-    }
-
-    @Test
-    void aSecondAskAfterTheFirstClosesStampsWhateverOwnerIsOpenAtThatLaterMoment() {
-        rendezvous.open(W, Rendezvous.Owner.of("term_lead"));
-        Rendezvous.AskTicket first = rendezvous.openAsk(W);
-        rendezvous.closeAsk(first.turnId());
-
-        // The forward waiter is reopened under a different owner before the second ask — mirrors
-        // answer() reopening with the owner it already checked, which can differ turn to turn.
-        rendezvous.close(W, rendezvous.currentWaiter(W));
-        rendezvous.open(W, Rendezvous.Owner.of("term_other"));
-
-        Rendezvous.AskTicket second = rendezvous.openAsk(W);
-        assertTrue(second.fresh());
-        assertEquals(Rendezvous.Owner.of("term_other"), rendezvous.askOwner(second.turnId()),
-                "a freshly-opened ask always copies whatever owner is open right now, not a stale one");
-    }
-
-    @Test
-    void askOwnerIsNullForAnUnknownOrLapsedTurn() {
-        assertNull(rendezvous.askOwner("no-such#1"));
-        Rendezvous.AskTicket t = rendezvous.openAsk(W);
-        rendezvous.closeAsk(t.turnId());
-        assertNull(rendezvous.askOwner(t.turnId()), "a closed ask no longer reports an owner");
-    }
-
-    @Test
-    void ownerPermitsIsFailClosedOnARecordAndThreeStatesAreDistinct() {
-        assertFalse(Rendezvous.Owner.permits(null, null),
-                "no owner on record refuses even a caller with no terminal");
-        assertFalse(Rendezvous.Owner.permits(null, "term_a"),
-                "no owner on record refuses a terminal-bearing caller too");
-        assertTrue(Rendezvous.Owner.permits(Rendezvous.Owner.UNNAMED_PRIMARY, null),
-                "the unnamed primary owner matches a caller with no terminal");
-        assertFalse(Rendezvous.Owner.permits(Rendezvous.Owner.UNNAMED_PRIMARY, "term_a"),
-                "the unnamed primary owner does not match a terminal-bearing caller");
-        assertTrue(Rendezvous.Owner.permits(Rendezvous.Owner.of("term_a"), "term_a"),
-                "a named owner matches the same terminal");
-        assertFalse(Rendezvous.Owner.permits(Rendezvous.Owner.of("term_a"), "term_b"),
-                "a named owner refuses a different terminal");
-        assertFalse(Rendezvous.Owner.permits(Rendezvous.Owner.of("term_a"), null),
-                "a named owner refuses the unnamed primary");
-    }
 }
@@ -343,7 +343,7 @@ class FleetAppAuthTest {

            // Clean up the still-open ask so the background thread does not linger past the test.
            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                    () -> messages.answer(turnId, "config.yaml", 5000, "term_a"));
+                    () -> messages.answer(turnId, "config.yaml", 5000));
            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
            deadline = System.currentTimeMillis() + 3000;
            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
@@ -358,154 +358,6 @@ class FleetAppAuthTest {
        }
    }

-    /**
-     * {@code POST /sessions/{id}/message} with a {@code turnId} refuses a caller whose terminal
-     * did not open the turn, even when that caller otherwise holds ANSWER rights, and leaves the
-     * turn open for the real owner to resolve. Covers the REST adapter's ANSWER gate for an
-     * async-send ({@code wait:false}) delegation.
-     */
-    @Test
-    void restAnswerIsRefusedForADifferentCallerOnAnAsyncSendDelegationButTheRealOwnerSucceeds() throws Exception {
-        FakeHerdr herdr = new FakeHerdr();
-        AgentControl agents = new AgentControl(herdr);
-        Injector injector = new Injector(agents);
-        Rendezvous rendezvous = new Rendezvous();
-        MessageService messages = new MessageService(agents, injector, rendezvous);
-
-        Javalin ownerApp = startOnSharedService(messages, herdr, FakeHerdr.WORKER_PID, Map.of("term_a", "lead-owner"));
-        Javalin attackerApp = startOnSharedService(messages, herdr, 9001L, Map.of("term_shell", "lead-attacker"));
-        try {
-            ObjectMapper mapper = new ObjectMapper();
-            HttpResponse<String> created = send(ownerApp.port(), "POST", "/sessions/term_target/message",
-                    "{\"content\":\"long task\",\"wait\":false}", null);
-            assertEquals(202, created.statusCode(), created.body());
-            String ticket = mapper.readTree(created.body()).path("ticket").asText(null);
-            assertNotNull(ticket, "the accepted response carried no ticket: " + created.body());
-
-            long deadline = System.currentTimeMillis() + 3000;
-            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
-                Thread.sleep(5);
-            }
-            assertTrue(rendezvous.isWaiting("term_target"), "the async send should have opened its rendezvous waiter");
-
-            CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
-                    () -> messages.ask("term_target", "which config file?", 5000));
-
-            MessageService.TaskView asking;
-            deadline = System.currentTimeMillis() + 3000;
-            do {
-                asking = messages.poll(ticket, null);
-                Thread.sleep(5);
-            } while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline);
-            assertEquals(MessageService.Phase.ASKING, asking.phase());
-            String turnId = asking.turnId();
-
-            HttpResponse<String> hijacked = send(attackerApp.port(), "POST", "/sessions/term_target/message",
-                    "{\"content\":\"hijack\",\"turnId\":\"" + turnId + "\"}", null);
-            assertEquals(403, hijacked.statusCode(), hijacked.body());
-            assertTrue(hijacked.body().contains("not_turn_owner"),
-                    "a different caller's answer must be refused as not_turn_owner: " + hijacked.body());
-            assertEquals("term_target", rendezvous.askSession(turnId),
-                    "a refused answer must leave the ask turn open");
-
-            CompletableFuture<HttpResponse<String>> owned = CompletableFuture.supplyAsync(() -> {
-                try {
-                    return send(ownerApp.port(), "POST", "/sessions/term_target/message",
-                            "{\"content\":\"config.yaml\",\"turnId\":\"" + turnId + "\"}", null);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            });
-            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-            deadline = System.currentTimeMillis() + 3000;
-            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
-                Thread.sleep(5);
-            }
-            assertTrue(rendezvous.resolve("term_target", "done"));
-            HttpResponse<String> ownedResponse = owned.get(5, TimeUnit.SECONDS);
-            assertEquals(200, ownedResponse.statusCode(), ownedResponse.body());
-            assertEquals("done", mapper.readTree(ownedResponse.body()).path("reply").asText(null),
-                    "the real owner's answer must resolve the worker's turn");
-        } finally {
-            ownerApp.stop();
-            attackerApp.stop();
-        }
-    }
-
-    /**
-     * As above, but the delegation is a blocking send ({@code wait:true}) instead of a ticket —
-     * the owning caller's own HTTP request is the one that surfaces the worker's question and
-     * later carries the real answer. Covers the REST adapter's ANSWER gate for a blocking-send
-     * delegation.
-     */
-    @Test
-    void restAnswerIsRefusedForADifferentCallerOnABlockingSendDelegationButTheRealOwnerSucceeds() throws Exception {
-        FakeHerdr herdr = new FakeHerdr();
-        AgentControl agents = new AgentControl(herdr);
-        Injector injector = new Injector(agents);
-        Rendezvous rendezvous = new Rendezvous();
-        MessageService messages = new MessageService(agents, injector, rendezvous);
-
-        Javalin ownerApp = startOnSharedService(messages, herdr, FakeHerdr.WORKER_PID, Map.of("term_a", "lead-owner"));
-        Javalin attackerApp = startOnSharedService(messages, herdr, 9001L, Map.of("term_shell", "lead-attacker"));
-        try {
-            ObjectMapper mapper = new ObjectMapper();
-            CompletableFuture<HttpResponse<String>> blocking = CompletableFuture.supplyAsync(() -> {
-                try {
-                    return send(ownerApp.port(), "POST", "/sessions/term_target/message",
-                            "{\"content\":\"long task\",\"wait\":true,\"timeoutMs\":5000}", null);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            });
-
-            long deadline = System.currentTimeMillis() + 3000;
-            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
-                Thread.sleep(5);
-            }
-            assertTrue(rendezvous.isWaiting("term_target"), "the blocking send should have opened its rendezvous waiter");
-
-            CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
-                    () -> messages.ask("term_target", "which config file?", 5000));
-
-            HttpResponse<String> questionResponse = blocking.get(5, TimeUnit.SECONDS);
-            assertEquals(202, questionResponse.statusCode(), questionResponse.body());
-            JsonNode question = mapper.readTree(questionResponse.body());
-            assertEquals("question", question.get("status").asText());
-            String turnId = question.get("turnId").asText();
-
-            HttpResponse<String> hijacked = send(attackerApp.port(), "POST", "/sessions/term_target/message",
-                    "{\"content\":\"hijack\",\"turnId\":\"" + turnId + "\"}", null);
-            assertEquals(403, hijacked.statusCode(), hijacked.body());
-            assertTrue(hijacked.body().contains("not_turn_owner"),
-                    "a different caller's answer must be refused as not_turn_owner: " + hijacked.body());
-            assertEquals("term_target", rendezvous.askSession(turnId),
-                    "a refused answer must leave the ask turn open");
-
-            CompletableFuture<HttpResponse<String>> owned = CompletableFuture.supplyAsync(() -> {
-                try {
-                    return send(ownerApp.port(), "POST", "/sessions/term_target/message",
-                            "{\"content\":\"config.yaml\",\"turnId\":\"" + turnId + "\"}", null);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            });
-            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-            deadline = System.currentTimeMillis() + 3000;
-            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
-                Thread.sleep(5);
-            }
-            assertTrue(rendezvous.resolve("term_target", "done"));
-            HttpResponse<String> ownedResponse = owned.get(5, TimeUnit.SECONDS);
-            assertEquals(200, ownedResponse.statusCode(), ownedResponse.body());
-            assertEquals("done", mapper.readTree(ownedResponse.body()).path("reply").asText(null),
-                    "the real owner's answer must resolve the worker's turn");
-        } finally {
-            ownerApp.stop();
-            attackerApp.stop();
-        }
-    }
-
    /**
     * As {@link #start}, but shares {@code messages} and {@code herdr} across several app
     * instances bound to different pids, each returned as its own started {@link Javalin} rather
@@ -2436,4 +2436,200 @@ class SessionManagerTest {
            return "threw:" + e.getClass().getName() + ":" + e.getMessage();
        }
    }
+
+    // --- fleetd #702: spawnedMemberRole must still answer for a pane mid-teardown ----------------
+
+    /**
+     * A {@link Worktrees} test double whose {@code hasUncommitted} runs an injected hook before
+     * answering. This is what lets a test resolve a releasing pane's terminal from inside the
+     * window {@link SessionManager#release} opens between removing the registry entry and
+     * actually stopping the pane — the hook runs synchronously on the release call's own thread,
+     * at the exact point {@code release} shells out to {@code git status}, so there is no sleep
+     * and no race to land in it.
+     */
+    private static final class HookedWorktrees implements Worktrees {
+        private Runnable hook;
+        private boolean dirty = false;
+
+        HookedWorktrees onHasUncommitted(Runnable hook) {
+            this.hook = hook;
+            return this;
+        }
+
+        @Override
+        public String add(String repoRoot, String branch, String baseRef) {
+            return "/wt/" + branch.replace('/', '_');
+        }
+
+        @Override
+        public void remove(String repoRoot, String worktreePath) {
+        }
+
+        @Override
+        public void deleteBranch(String repoRoot, String branch) {
+        }
+
+        @Override
+        public boolean hasUncommitted(String worktreePath) {
+            if (hook != null) {
+                hook.run();
+            }
+            return dirty;
+        }
+
+        @Override
+        public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+        }
+
+        @Override
+        public String repoRoot(String cwd) {
+            return "/repo";
+        }
+
+        @Override
+        public java.util.Optional<String> snapshot(String worktreePath, String branch, String message) {
+            return java.util.Optional.empty();
+        }
+
+        @Override
+        public WipRefStats wipRefs(String repoRoot) {
+            return new WipRefStats(0, 0L);
+        }
+
+        @Override
+        public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+            return 0;
+        }
+
+        @Override
+        public void shareWithGroup(String repoRoot, String worktreePath) {
+        }
+    }
+
+    @Test
+    void spawnedMemberRoleResolvesTheLiveRegisteredRole() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller", null);
+
+        assertEquals(s.role(), sessions.spawnedMemberRole(s.terminalId()),
+                "a registered session resolves to its own role");
+        assertNull(sessions.spawnedMemberRole("term_unknown"),
+                "a terminal with no session at all resolves to null");
+    }
+
+    @Test
+    void spawnedMemberRoleIsNullOnceReleaseFullyCompletes() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr, new HookedWorktrees());
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702a", null));
+
+        sessions.release(s.paneId());
+
+        assertNull(sessions.spawnedMemberRole(s.terminalId()),
+                "once release has fully finished, the terminal is neither registered nor releasing");
+    }
+
+    @Test
+    void spawnedMemberRoleStillAnswersBetweenTheRegistryRemovalAndThePaneStop() {
+        FakeHerdr herdr = new FakeHerdr();
+        java.util.concurrent.atomic.AtomicReference<MemberRole> duringWindow = new java.util.concurrent.atomic.AtomicReference<>();
+        HookedWorktrees worktrees = new HookedWorktrees();
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702b", null));
+        worktrees.onHasUncommitted(() -> {
+            // This runs from INSIDE release()'s git-status shell-out: the registry entry is
+            // already gone, but the pane has not stopped yet — a real call landing inside the
+            // exact window fleetd #702 reports, so no sleep and no race is needed to reach it.
+            assertTrue(sessions.get(s.paneId()).isEmpty(),
+                    "sanity: the registry entry is already gone at this point");
+            duringWindow.set(sessions.spawnedMemberRole(s.terminalId()));
+        });
+
+        sessions.release(s.paneId());
+
+        assertEquals(s.role(), duringWindow.get(),
+                "spawnedMemberRole must still answer the live role while the pane is mid-teardown, "
+                        + "not only while the session is still in the registry");
+        assertNull(sessions.spawnedMemberRole(s.terminalId()),
+                "and once release has fully finished, the window is closed too");
+    }
+
+    /**
+     * {@code Releasing} is private, so its {@code enter}/{@code leave} are reached through
+     * reflection. Each is called directly, never through {@link SessionManager#release} or
+     * {@link SessionManager#spawnedMemberRole}, so a test naming one of them exercises only that
+     * one — a regression in the other can never hide behind it.
+     */
+    private static Object releasingOf(int depth, String terminalId, MemberRole role) throws ReflectiveOperationException {
+        Class<?> cls = Class.forName("dev.ltms.fleet.session.SessionManager$Releasing");
+        java.lang.reflect.Constructor<?> ctor = cls.getDeclaredConstructor(int.class, String.class, MemberRole.class);
+        ctor.setAccessible(true);
+        return ctor.newInstance(depth, terminalId, role);
+    }
+
+    private static Object releasingEnter(Object prior, MemberSession known) throws ReflectiveOperationException {
+        Class<?> cls = Class.forName("dev.ltms.fleet.session.SessionManager$Releasing");
+        java.lang.reflect.Method m = cls.getDeclaredMethod("enter", cls, MemberSession.class);
+        m.setAccessible(true);
+        return m.invoke(null, prior, known);
+    }
+
+    private static Object releasingLeave(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("leave");
+        m.setAccessible(true);
+        return m.invoke(releasing);
+    }
+
+    private static int depthOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("depth");
+        m.setAccessible(true);
+        return (int) m.invoke(releasing);
+    }
+
+    private static String terminalIdOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("terminalId");
+        m.setAccessible(true);
+        return (String) m.invoke(releasing);
+    }
+
+    private static MemberRole roleOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("role");
+        m.setAccessible(true);
+        return (MemberRole) m.invoke(releasing);
+    }
+
+    @Test
+    void releasingLeaveStepsDownADepthGreaterThanOneInsteadOfRemovingIt() throws ReflectiveOperationException {
+        Object depthTwo = releasingOf(2, "term_a", MemberRole.DEV);
+
+        Object afterLeave = releasingLeave(depthTwo);
+
+        assertNotNull(afterLeave,
+                "depth 2 means another release of the SAME pane is still mid-teardown; leave() must "
+                        + "step the depth down, never remove the marker outright — removing it here "
+                        + "is what a plain Set would do, and would reopen the window the still-in-"
+                        + "flight release is relying on staying closed");
+        assertEquals(1, depthOf(afterLeave));
+        assertEquals("term_a", terminalIdOf(afterLeave));
+        assertEquals(MemberRole.DEV, roleOf(afterLeave));
+    }
+
+    @Test
+    void releasingEnterPreservesThePriorTerminalWhenTheOverlappingCallHasNoSessionOfItsOwn()
+            throws ReflectiveOperationException {
+        Object prior = releasingOf(1, "term_a", MemberRole.DEV);
+
+        Object afterEnter = releasingEnter(prior, null);
+
+        assertEquals(2, depthOf(afterEnter), "depth still increments whether or not this enter knows its session");
+        assertEquals("term_a", terminalIdOf(afterEnter),
+                "a losing CAS (the releaseIfCurrent race fleetd #702 is about) has no session of "
+                        + "its own to pass as known, and must not blank out the terminal the winning "
+                        + "call already recorded — that terminal is what spawnedMemberRole matches "
+                        + "against");
+        assertEquals(MemberRole.DEV, roleOf(afterEnter));
+    }
 }