#324: read task.turnId once in finishAsyncTask to stop an NPE from ask()'s unlocked forgetting

answer() holds sessionLocks while finishAsyncTask reads the volatile Task.turnId twice — once to
check it is non-null, once as the ConcurrentHashMap.remove key. ask()'s own timeout path mutates
the same field with no lock, via clearAsyncQuestion(turnId, true). volatile makes each read fresh
but not the pair atomic, so the field can go null between the two reads and remove(null, task)
throws NullPointerException on the lead's own answer() call, even though the reply already
completed on the line above.

Capture task.turnId into a local once and use that for both the check and the removal.

Added a package-private test seam (finishAsyncTaskRaceHook + forgetTurnForTest) so a test can force
the exact interleaving deterministically, by running the identical clearAsyncQuestion(turnId, true)
cleanup ask() uses, at the point between finishAsyncTask's former two reads. Both are inert (null)
in production.

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

@@ -1230,14 +1230,61 @@ public final class MessageService {
         }
     }
 
-    /** Complete and detach an async ticket after its worker's actual terminal reply. */
+    /**
+     * Complete and detach an async ticket after its worker's actual terminal reply.
+     *
+     * <p><strong>fleetd #324.</strong> {@code task.turnId} is read into {@code turnId} exactly once.
+     * It used to be read twice — once for the null check, once as the removal key — and {@code
+     * volatile} makes each of those reads individually fresh but does not make the pair atomic.
+     * {@link #answer} calls this while holding {@code sessionLocks} for the target; {@link #ask}'s
+     * own timeout path calls {@link #clearAsyncQuestion} (which nulls {@link Task#turnId}) under no
+     * lock at all. When that unlocked null-out landed between the two reads here, the second read saw
+     * {@code null} and {@code asyncTasksByTurn.remove(null, task)} threw {@code NullPointerException}
+     * on the lead's own {@code answer()} call — even though {@code task.future.complete(result)} on
+     * the line above had already run, so the answer was in fact delivered. Capturing the field once
+     * removes the torn read; see the ticket for why the wider asymmetry between the locked and
+     * unlocked sides is not fixed by this alone.
+     */
     private void finishAsyncTask(Task task, Reply result) {
         task.future.complete(result);
-        if (task.turnId != null) {
-            asyncTasksByTurn.remove(task.turnId, task);
+        String turnId = task.turnId;
+        if (turnId != null) {
+            if (finishAsyncTaskRaceHook != null) {
+                // Test-only (fleetd #324): see the field's own javadoc.
+                finishAsyncTaskRaceHook.run();
+            }
+            asyncTasksByTurn.remove(turnId, task);
         }
     }
 
+    /**
+     * Null in production; test seam for fleetd #324 — invoked from {@link #finishAsyncTask(Task,
+     * Reply)} right after {@code task.turnId}'s null-check passes and before the (now-local) value is
+     * used for the removal. A test installs this to force, deterministically, the exact interleaving
+     * that a real race between this method and {@link #ask}'s unlocked timeout cleanup can otherwise
+     * only produce by chance: firing it here reproduces "the field went null between the check and the
+     * use" against the pre-fix code, and demonstrates the fix tolerates it (the captured local is used
+     * unconditionally, so a hook that nulls the field afterward cannot affect this call).
+     */
+    private volatile Runnable finishAsyncTaskRaceHook;
+
+    /**
+     * Test-only (fleetd #324): install {@link #finishAsyncTaskRaceHook}. Package-private so the test,
+     * in the same package, can reach it without widening any production API.
+     */
+    void setFinishAsyncTaskRaceHookForTest(Runnable hook) {
+        this.finishAsyncTaskRaceHook = hook;
+    }
+
+    /**
+     * Test-only (fleetd #324): run the exact production cleanup {@link #ask}'s own timeout path runs
+     * unlocked — {@link #clearAsyncQuestion(String, boolean)} with {@code forgetTurn=true} — so a test
+     * can reproduce that specific mutation instead of hand-rolling an approximation of it.
+     */
+    void forgetTurnForTest(String turnId) {
+        clearAsyncQuestion(turnId, true);
+    }
+
     /** Complete the async ticket correlated to a specific answered turn. */
     private void finishAsyncTask(String turnId, Reply result) {
         Task task = asyncTasksByTurn.get(turnId);

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

@@ -940,6 +940,54 @@ class MessageServiceTest {
         assertEquals("PR opened: https://example/pulls/42", view.reply());
     }
 
+    /**
+     * fleetd #324: {@code answer()} holds {@code sessionLocks} for the target and, once the worker's
+     * real terminal reply arrives, calls {@code finishAsyncTask}, which used to read the volatile
+     * {@code task.turnId} twice — once to check it is non-null, once as the key for
+     * {@code asyncTasksByTurn.remove}. {@code ask()}'s own timeout path mutates the same field with no
+     * lock at all. This test does not wait for a real race to land in that narrow window between the
+     * two reads — instead it drives the exact sequence the ticket describes (worker asks, primary
+     * answers, worker's real reply arrives) and, via a package-private test hook wired to fire at
+     * precisely that point, runs the identical production cleanup {@code ask()}'s timeout catch block
+     * runs ({@code clearAsyncQuestion(turnId, true)}) so the field goes {@code null} between the two
+     * reads deterministically rather than by chance.
+     *
+     * <p>What this proves: given that exact interleaving, {@code answer()} must not throw and the
+     * ticket must still resolve to the worker's real reply. What it does not prove: that the
+     * interleaving itself is reachable in production — that is established by reading the code (see
+     * the ticket), not by this test, since forcing it via a hook is not the same as two independent
+     * threads racing on their own schedules.
+     */
+    @Test
+    void finishAsyncTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+        String turnId = asking.turnId();
+
+        // Fire ask()'s own unlocked timeout cleanup at the moment finishAsyncTask has already checked
+        // task.turnId is non-null but has not yet used it — the exact torn-read window fleetd #324
+        // describes.
+        messages.setFinishAsyncTaskRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
+
+        CompletableFuture<MessageService.Reply> answer =
+                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
+
+        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
+                "the lead's own answer() call must not throw because ask()'s timeout cleanup raced it");
+        MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+        assertEquals("PR opened: https://example/pulls/42", done.reply(),
+                "the ticket must still resolve to the worker's real reply despite the forced race");
+    }
+
     @Test
     void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
         String ticket = messages.sendAsync(T, "task that asks");