Merge #572: answer()'s session-lock release is pinned on all four exits

fleetd #572. MessageService releases its per-session lock in a finally at two sites. Removing the
send() one failed 22 tests and errored 1. Removing the answer() one left the whole suite green.
Skip that unlock and the thread holds the session lock forever, so every later send or answer to
that session blocks permanently — no exception, no log line.

Tests only. Against current main the diff is one file, MessageServiceTest.java, +178 lines.

Four new tests, one per exit of answer(): normal REPLIED, TIMED_OUT_WORKING, the ExecutionException
rethrow, and the InterruptedException rethrow. Each proves REACQUISITION rather than the return
value — a bounded follow-up send on the SAME session must not come back BUSY, and send reports BUSY
only when tryLock itself timed out, so a non-BUSY probe is specifically evidence the lock was free.

Lead verification, re-running rather than accepting the worker's numbers, on the branch merged with
main at 3f8c38f:
  baseline   -> exit 0, 1765 tests from Maven and an independent sum over 131 reports (1761 + 4)
  mutation L -> BUILD FAILURE, exactly 4 failures, all four new tests by name

THE TRAP THE WORKER CAUGHT ITSELF, which is the most valuable thing in this PR. Their first draft of
the TIMED_OUT_WORKING test called answer() inline on the test thread and then probed with send() on
that same thread. It FALSE-PASSED under the mutation: lock is a ReentrantLock, so the same thread
reenters for free whether or not unlock() ran. A same-thread probe proves reentrancy, not release.
They found it only because they actually ran the mutation instead of trusting a green baseline, moved
answer() onto a background thread, and re-proved the kill. The pitfall is now documented in the test's
own javadoc, with the measurement that exposed it.

NOTE FOR ANYONE RE-RUNNING THE TICKET'S RECIPE: the ticket records `sed '1218s|...'`, measured before
#551 merged. #551 edited MessageService.java, and the line is now 1231. Locate it fresh — a
line-anchored sed against a stale number mutates the wrong line and reports a meaningless green. The
anchor count is the check that catches this: `grep -Fxc '            lock.unlock();'` must read 2
pristine and 1 after.

Shape reported by the worker, not investigated and not fixed: the same two-line cleanup
`asyncTasksByWaiter.remove(reply); rendezvous.close(...)` appears at three sites — send()'s finally,
answer()'s inner finally, and answer()'s inline STALE_TURN early return, which is NOT inside a
finally. Same maintained-at-N-sites shape as this finding. Filed separately.

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

@@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -579,6 +580,183 @@ class MessageServiceTest {
         assertEquals("config.yaml", a.answer());
     }
 
+    // --- fleetd #572: answer() must release the session lock on EVERY exit, not just return it -----
+    //
+    // answer()'s session lock is released in an outer `finally` (MessageService.java:1217-1219) that
+    // wraps its whole body. Removing that one line survives the entire suite: coverage is not the
+    // gap, assertion is — every existing answer() test checks the RETURN VALUE, never that the lock
+    // it took is reacquirable afterward. If it is not, the session is wedged forever with no
+    // exception and no log line. Each test below drives answer() through one of its four exits
+    // (normal reply, TIMED_OUT_WORKING, ExecutionException, InterruptedException) and then proves
+    // reacquisition the only way that actually proves it: a bounded follow-up `send` on the SAME
+    // session must not come back BUSY. A `send` only reports BUSY when `tryLock` itself timed out
+    // (MessageService.java:926-928) — every other early return in `send` still passes through its own
+    // lock-acquired `try`, so a non-BUSY probe result is specifically evidence the lock was free.
+
+    /** The REPLIED exit (the happy path) — the resumed worker's real {@code fleet_reply} arrives. */
+    @Test
+    void answerReleasesTheSessionLockAfterANormalReply() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+
+        CompletableFuture<MessageService.Reply> answer =
+                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
+        assertTrue(rendezvous.resolve(T, "done"), "the worker's final reply resolves the answering send");
+        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);
+        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");
+    }
+
+    /**
+     * The {@code TIMED_OUT_WORKING} exit. Same setup as {@link #answerTimesOutWhenTheResumedWorkerNeverReplies}
+     * (which only checks {@code answer()}'s return value — exactly the assertion the fleetd #572
+     * mutation survives), plus the reacquisition proof that test does not make.
+     *
+     * <p>{@code answer()} MUST run on its own thread here, not the test's main thread: {@code lock}
+     * is a {@link java.util.concurrent.locks.ReentrantLock}, so a probe {@code send} issued by the
+     * SAME thread that (under the mutation) still "holds" it would reenter for free and report a
+     * false pass — reentrancy, not release. Measured while writing this test: with {@code answer()}
+     * called inline, this test stayed green under the mutation while its three siblings correctly
+     * went red.
+     */
+    @Test
+    void answerReleasesTheSessionLockAfterATimedOutWorkingReturn() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+
+        // 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));
+        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);
+        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");
+    }
+
+    /**
+     * The {@code ExecutionException} exit. Forces it directly on answer()'s own reopened forward
+     * waiter — {@code rendezvous.currentWaiter(T)} is the exact {@code CompletableFuture} its
+     * {@code reply.get(...)} is blocked on — rather than trying to make a real worker fail, since the
+     * failure mode under test is in answer()'s own wait, not in how it got triggered.
+     */
+    @Test
+    void answerReleasesTheSessionLockWhenTheReplyFutureFailsExceptionally() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+        String turnId = q.turnId();
+
+        java.util.concurrent.atomic.AtomicReference<Throwable> caught =
+                new java.util.concurrent.atomic.AtomicReference<>();
+        Thread answerer = new Thread(() -> {
+            try {
+                messages.answer(turnId, "config.yaml", 5000);
+                caught.set(new AssertionError("expected answer() to throw"));
+            } catch (Throwable t) {
+                caught.set(t);
+            }
+        });
+        answerer.start();
+        awaitWaiting(); // answer() re-opened its forward waiter and is about to block on it
+
+        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
+        assertNotNull(waiter, "answer() should have registered a forward waiter for the worker session");
+        waiter.completeExceptionally(new RuntimeException("boom"));
+
+        answerer.join(5000);
+        assertFalse(answerer.isAlive(), "answer() should have left by throwing once its reply future failed");
+        Throwable thrown = caught.get();
+        assertNotNull(thrown, "answer() should have thrown");
+        assertTrue(thrown instanceof RuntimeException, "a RuntimeException cause is rethrown as-is: " + thrown);
+        assertEquals("boom", thrown.getMessage());
+
+        ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer
+
+        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");
+    }
+
+    /**
+     * The {@code InterruptedException} exit. Same shape as the {@code ExecutionException} case above,
+     * but the thread blocked in {@code reply.get(...)} is interrupted instead of the future failing.
+     */
+    @Test
+    void answerReleasesTheSessionLockWhenInterrupted() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+        String turnId = q.turnId();
+
+        java.util.concurrent.atomic.AtomicReference<Throwable> caught =
+                new java.util.concurrent.atomic.AtomicReference<>();
+        Thread answerer = new Thread(() -> {
+            try {
+                messages.answer(turnId, "config.yaml", 5000);
+                caught.set(new AssertionError("expected answer() to throw"));
+            } catch (Throwable t) {
+                caught.set(t);
+            }
+        });
+        answerer.start();
+        awaitWaiting(); // answer() re-opened its forward waiter and is about to block on it
+
+        answerer.interrupt();
+        answerer.join(5000);
+        assertFalse(answerer.isAlive(), "answer() should have left by throwing once its thread was interrupted");
+        Throwable thrown = caught.get();
+        assertNotNull(thrown, "answer() should have thrown");
+        assertTrue(thrown instanceof IllegalStateException,
+                "an interrupted wait is wrapped in IllegalStateException: " + thrown);
+
+        ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer
+
+        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");
+    }
+
     @Test
     void pollReturnsNullForAnUnknownTicket() {
         assertNull(messages.poll("task-999999"), "a ticket that was never minted is unknown");