fleetd#329: fix silent async-ticket bugs in MessageService (F1/F2/F3)

F2 (sendAsync executor catch): log when completeExceptionally returns
false, so an exception thrown after finishAsyncTask already completed
the ticket's future is no longer silently lost.

F1 (answer()'s stranded async ticket): reuse the Task reference answer()
already looked up before rendezvous.answerAsk(), instead of a second
asyncTasksByTurn lookup by turnId in finishAsyncTask. The second lookup
raced ask()'s unlocked timeout cleanup, which could forget turnId first
and leave the ticket stuck PENDING even though answer() itself returned
REPLIED. The #282 chained-ask guard is unaffected: it is still keyed on
result.outcome() == QUESTION, not on this lookup. Removed the now-unused
finishAsyncTask(String, Reply) overload.

F3 (reply()'s orphan recovery path): read orphan.turnId once instead of
twice, closing the same double-read shape fleetd #324 fixed in
finishAsyncTask.

Each fix has its own test plus a test-only race hook (mirroring #324's
finishAsyncTaskRaceHook) to force the exact interleaving deterministically.
Mutation-tested each fix by reverting it, confirming the real failure
(swallowed exception / PENDING ticket / NullPointerException), then
restoring it.

mvn clean install: Tests run: 1345, Failures: 0, Errors: 0, Skipped: 0,
BUILD SUCCESS.

fleetd/src/main/java/dev/ltms/fleet/config/ConfigRef.java

@@ -45,8 +45,12 @@ import java.util.function.Supplier;
  *       {@link #changedDeferredKeys}), {@code primary:} (fleetd #326 — {@code Fleetd.java:506, 519,
  *       520} read {@code cfg.primary()} only off the startup snapshot to build {@code
  *       PrimaryRegistry} and size {@code ReplyPushLoop}'s reminder cap/backoff, and neither is
- *       rebuilt on reload; a lead whose pinned terminal changed under a running daemon stays
- *       unresolved as primary until a restart), {@code configReload:} (fleetd #326 — {@code
+ *       rebuilt on reload. Say the consequence exactly: {@code primary.terminal} is DEPRECATED
+ *       (CB-532, and {@code Fleetd.java:511} warns about it at startup) — a lead's identity comes
+ *       from {@code leaders:}/{@code leadScan:}, so changing this pin does not demote or promote a
+ *       lead that uses those. What a changed pin still does not take effect on until a restart is
+ *       the fallback nudge destination the pin remains, the deprecated identity path for an operator
+ *       who still relies on it, and {@code pushReminders}/{@code pushBackoffMs}), {@code configReload:} (fleetd #326 — {@code
  *       Fleetd.java:679-680} read it only at startup to decide whether to build a {@code
  *       ConfigWatcher} at all and with what interval; the watcher that would apply a later change is
  *       itself built once, so a running watcher keeps polling on its original enabled flag and
@@ -73,6 +77,24 @@ import java.util.function.Supplier;
  *       listening.</li>
  * </ul>
  *
+ * <p><strong>The denominator, measured on 2026-09-04 (fleetd #326).</strong> {@code FleetConfig} has
+ * 22 top-level record components. Four of them are named nowhere in this file, and the reason
+ * differs per key, so do not read "absent" as "forgotten":
+ * <ul>
+ *   <li>{@code memberCredentials} and {@code memberLoginShell} are <strong>hot</strong> and
+ *       correctly absent — both are read live off {@code config.get()} at spawn time
+ *       ({@code Fleetd.java:198, 205, 729} and {@code HerdrPeerLauncher#configuredMemberLoginShell}),
+ *       so a reload takes effect on the next spawn with no entry needed here.</li>
+ *   <li>{@code health} and {@code coordinator} are <strong>undecided</strong>, not hot. Each is read
+ *       both ways at different sites, so neither fits the three classes above as a whole key. Until
+ *       that is settled a reload touching them reports a bare "config reloaded", which under-claims.
+ *       Deciding it is what a top-level coverage checker (the {@link ConfigRefProfileCoverageTest}
+ *       shape, one level up) is waiting on — without it, such a checker cannot express the answer.</li>
+ * </ul>
+ * The point of writing the count down: "not mentioned in this file" looks identical for a key that
+ * is correctly hot and for a key nobody triaged. Twice now — {@code worktreeGroup} (#323) and
+ * {@code primary}/{@code configReload} (#326) — the second kind hid among the first.
+ *
  * <p><strong>A cold change refuses the whole reload.</strong> Not the hot half applied and the cold
  * half warned about: that would leave the running daemon in a state matching no file on disk, which
  * is the worst thing a reload can do to an operator debugging one. Refusing keeps the invariant that
@@ -246,7 +268,9 @@ public final class ConfigRef implements Supplier<FleetConfig> {
         }
         // fleetd #326: Fleetd.java:506, 519, 520 read cfg.primary() only off the startup snapshot
         // (PrimaryRegistry's pinned terminal, ReplyPushLoop's reminder cap and backoff) — neither is
-        // rebuilt on reload, so a changed pin needs a restart before a lead resolves as primary again.
+        // rebuilt on reload, so a changed value needs a restart. Note what it does NOT mean:
+        // primary.terminal is deprecated (CB-532), identity comes from leaders:/leadScan:, so a lead
+        // using those is unaffected by this pin either way. See the class doc for the exact scope.
         if (!Objects.equals(old.primary(), fresh.primary())) {
             changed.add("primary");
         }

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

@@ -466,8 +466,18 @@ public final class MessageService {
         if (candidates.size() == 1) {
             Task orphan = candidates.get(0);
             if (orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
-                if (orphan.turnId != null) {
-                    asyncTasksByTurn.remove(orphan.turnId, orphan);
+                // fleetd #329 (F3): read orphan.turnId once. It used to be read twice, under no
+                // lock — once for this null check, once as the removal key — the same double-read
+                // shape fleetd #324 fixed in finishAsyncTask. clearAsyncQuestion's unlocked
+                // forgetTurn=true path (ask()'s timeout cleanup) can null the field between the two
+                // reads; capturing it once removes the torn read here too.
+                String turnId = orphan.turnId;
+                if (turnId != null) {
+                    if (replyOrphanTurnIdRaceHookForTest != null) {
+                        // Test-only (fleetd #329, F3): see the field's own javadoc.
+                        replyOrphanTurnIdRaceHookForTest.run();
+                    }
+                    asyncTasksByTurn.remove(turnId, orphan);
                 }
                 count(FleetMetrics.REPLIES, "path", "async-recovered");
                 return true;                   // the ticket itself took it — no inbox stranding at all
@@ -1002,13 +1012,36 @@ public final class MessageService {
                 // Measured when #282 was merged: this guard is DEFENCE IN DEPTH, not the thing
                 // that makes the chained ask work. ask() calls markAsyncQuestion (:860) before
                 // resolveQuestion (:861), so by the time this thread wakes, the task has already
-                // moved to the new turnId and finishAsyncTask(oldTurnId, ...) finds nothing. Removing
-                // this guard alone leaves the test green. Keep it anyway: it mirrors sendAsync's
-                // sibling guard, and that sibling's own comment (:1017) warns the two orderings are
-                // not something to rely on. Do NOT delete it as dead code without re-checking that
-                // ordering, and do not treat it as the sole protection either.
-                if (result.outcome() != Outcome.QUESTION) {
-                    finishAsyncTask(turnId, result);
+                // moved to the new turnId — this outcome check, not a Task lookup, is what tells the
+                // two cases apart (see fleetd #329 below). Removing this guard alone leaves the test
+                // green. Keep it anyway: it mirrors sendAsync's sibling guard, and that sibling's own
+                // comment (:1017) warns the two orderings are not something to rely on. Do NOT delete
+                // it as dead code without re-checking that ordering, and do not treat it as the sole
+                // protection either.
+                //
+                // fleetd #329 (F1): complete the SAME Task object this method already looked up at
+                // :991, rather than re-resolving it from turnId a second time. The old
+                // finishAsyncTask(turnId, result) did its own asyncTasksByTurn.get(turnId) here, and
+                // that second lookup races ask()'s own timeout path: ask()'s ticket.answer().get(...)
+                // can time out (or lose that exact race) at essentially the same instant this method's
+                // rendezvous.answerAsk(turnId, ...) above already succeeded, running
+                // markAskTimedOut + clearAsyncQuestion(turnId, true) with no lock at all — which
+                // forgets turnId (removes it from asyncTasksByTurn, nulls Task.turnId) before this
+                // thread ever gets here. The worker's real reply then arrived, this method's own wait
+                // woke up with it, and the by-turnId lookup found nothing: the async ticket's future
+                // was never completed, so fleet_poll{ticket} stayed PENDING forever even though
+                // answer() itself correctly returned REPLIED. Reusing the reference captured at :991
+                // — before that race window opens — sidesteps the second lookup entirely: it is the
+                // identical Task whatever asyncTasksByTurn or Task.turnId say by the time we reach
+                // this point, so finishAsyncTask(Task, Reply) can still complete its future and detach
+                // it using whatever turnId it now reads. The #282 chained-ask case is unaffected
+                // because it is still gated purely by result.outcome() == QUESTION above, which does
+                // not depend on this lookup — widening what "task" means here cannot complete a ticket
+                // the chained ask deliberately left open. When task is null (this was never an async
+                // ticket — a blocking fleet_ask's answer() call has no Task at all), there is nothing
+                // to complete, matching the old lookup-miss behaviour.
+                if (result.outcome() != Outcome.QUESTION && task != null) {
+                    finishAsyncTask(task, result);
                 }
                 return result;
             } catch (TimeoutException e) {
@@ -1059,7 +1092,7 @@ public final class MessageService {
             // this fires exactly once, from whichever path completes it: finishAsyncTask(task, result)
             // below on any non-QUESTION outcome of send() — a worker's fleet_reply, the CB-106
             // completion fallback, a CB-109 wedge, TIMED_OUT, BUSY, or BACKEND_EXHAUSTED — the same
-            // finishAsyncTask reached via answer()'s finishAsyncTask(turnId, result) once a QUESTION
+            // finishAsyncTask reached via answer()'s own finishAsyncTask(task, result) once a QUESTION
             // is resolved, completeExceptionally(t) just below when send() itself throws, or a CB-516
             // abandon() on teardown. Without this, MessageService.reply's rendezvous fast path (the
             // one an async ticket always takes) never told the push loop anything happened — see the
@@ -1079,7 +1112,22 @@ public final class MessageService {
                     finishAsyncTask(task, result);
                 }
             } catch (Throwable t) {
-                task.future.completeExceptionally(t);
+                // fleetd #329 (F2): finishAsyncTask above already completes task.future — on its very
+                // first line — before doing anything else, so anything that throws afterward (inside
+                // finishAsyncTask's own cleanup, or from a future addition to this try block) lands
+                // here with the future already resolved. completeExceptionally on an already-completed
+                // future is a silent no-op: it returns false and does nothing, so without the check
+                // below the exception simply vanished — no log, no metric, nothing. Measured (see the
+                // ticket): temporarily reintroducing the fleetd #324 NPE reproduced 19 real exceptions
+                // on the ordinary path, with 74/74 tests staying green and not one log line produced.
+                // Do not stop completing the future first — finishAsyncTask completing before it
+                // cleans up is what makes a late failure harmless to the ticket's own result — only
+                // add the missing visibility for the case where that step, or whatever ran after it,
+                // has already lost the race to report through the future.
+                if (!task.future.completeExceptionally(t)) {
+                    log.error("async send {} -> {} threw after its ticket was already resolved",
+                            ticket, target, t);
+                }
             }
         });
         pruneTerminalTickets();
@@ -1230,20 +1278,106 @@ 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);
+        if (afterFinishAsyncTaskCompleteHookForTest != null) {
+            // Test-only (fleetd #329, F2): see the field's own javadoc.
+            afterFinishAsyncTaskCompleteHookForTest.run();
+        }
+        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);
         }
     }
 
-    /** Complete the async ticket correlated to a specific answered turn. */
-    private void finishAsyncTask(String turnId, Reply result) {
-        Task task = asyncTasksByTurn.get(turnId);
-        if (task != null) {
-            finishAsyncTask(task, result);
-        }
+    /**
+     * 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;
+    }
+
+    /**
+     * Null in production; test seam for fleetd #329 (F2) — invoked from {@link #finishAsyncTask(Task,
+     * Reply)} unconditionally, immediately after {@code task.future.complete(result)} runs (before
+     * {@link Task#turnId} is even read, so it fires regardless of whether this task was ever asked).
+     * A test installs this to force an exception into exactly the shape fleetd #329 identified:
+     * something throws inside {@link #sendAsync}'s executor task after the async ticket's future is
+     * already resolved, so the surrounding {@code catch (Throwable t)} can only report failure through
+     * {@code completeExceptionally} — a silent no-op on an already-completed future. Engineering a
+     * real exception to land in that exact post-completion window is what the ticket itself had to do
+     * by temporarily deleting a production guard (fleetd #324's {@code turnId != null} check); this
+     * hook drives the identical shape deterministically instead.
+     */
+    private volatile Runnable afterFinishAsyncTaskCompleteHookForTest;
+
+    /**
+     * Test-only (fleetd #329, F2): install {@link #afterFinishAsyncTaskCompleteHookForTest}.
+     * Package-private so the test, in the same package, can reach it without widening any production
+     * API.
+     */
+    void setAfterFinishAsyncTaskCompleteHookForTest(Runnable hook) {
+        this.afterFinishAsyncTaskCompleteHookForTest = hook;
+    }
+
+    /**
+     * Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
+     * the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
+     * value is used as the {@code asyncTasksByTurn} removal key. Mirrors {@link
+     * #finishAsyncTaskRaceHook} exactly, for the structurally identical double-read fleetd #324 fixed
+     * in {@link #finishAsyncTask}: a test installs this to force, deterministically, {@code
+     * clearAsyncQuestion}'s unlocked {@code forgetTurn=true} path nulling {@link Task#turnId} in that
+     * exact window, and to confirm the single-read fix tolerates it (the captured local is used
+     * unconditionally, so a hook that nulls the field afterward cannot affect this call).
+     */
+    private volatile Runnable replyOrphanTurnIdRaceHookForTest;
+
+    /**
+     * Test-only (fleetd #329, F3): install {@link #replyOrphanTurnIdRaceHookForTest}. Package-private
+     * so the test, in the same package, can reach it without widening any production API.
+     */
+    void setReplyOrphanTurnIdRaceHookForTest(Runnable hook) {
+        this.replyOrphanTurnIdRaceHookForTest = 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);
     }
 
     /** A new send must not open a waiter while an async ticket owns this worker's paused turn. */

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

@@ -1,5 +1,9 @@
 package dev.ltms.fleet.msg;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.AgentStatus;
 import dev.ltms.fleet.herdr.FakeHerdr;
@@ -11,6 +15,7 @@ import dev.ltms.fleet.mcp.PrimaryRegistry;
 import dev.ltms.fleet.inject.Injector;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -940,6 +945,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");
@@ -993,6 +1046,105 @@ class MessageServiceTest {
                 "the reply completed its own ticket directly and never touched the inbox");
     }
 
+    /**
+     * fleetd #329 (F1). {@code answer()} completes the async ticket by looking {@code turnId} up in
+     * {@code asyncTasksByTurn} a SECOND time (the first is at :991, purely to re-register the
+     * {@code asyncTasksByWaiter} entry for #282's chained-ask case). That second lookup races
+     * {@code ask()}'s own unlocked timeout cleanup ({@code clearAsyncQuestion(turnId, true)}, run from
+     * {@code markAskTimedOut} + forgetting): {@code ask()}'s {@code ticket.answer().get(timeoutMillis)}
+     * can time out at essentially the same instant {@code answer()}'s {@code rendezvous.answerAsk}
+     * call above already succeeded and unblocked the worker. fleetd #324's single-read fix does not
+     * help here — that fixed a torn read of one already-held {@link MessageService} internal
+     * {@code Task}; this is a second, independent map lookup by a caller that no longer holds the
+     * {@code Task} it already found once.
+     *
+     * <p>This test does not wait for that real race to land on its own schedule — it drives the exact
+     * sequence the ticket describes (worker asks, primary answers, worker's real reply arrives) and
+     * fires the identical production cleanup {@code ask()}'s timeout path runs
+     * ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd #324
+     * already merged) at the point between "the worker's own {@code ask()} call has unblocked" and
+     * "the worker's real {@code fleet_reply} arrives" — the exact window fleetd #329 names.
+     *
+     * <p>What this proves: given that exact interleaving, the async ticket must still resolve to the
+     * worker's real reply, not stay stuck {@link MessageService.Phase#PENDING} forever. What it does
+     * not prove: that the interleaving itself is reachable in production on its own timing — that is
+     * established by reading the code (see the ticket's "path in"), not by this test, for the same
+     * reason fleetd #324's own race test says so.
+     */
+    @Test
+    void aReplyRacingAsksTimeoutCleanupStillCompletesTheAsyncTicket() 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();
+
+        CompletableFuture<MessageService.Reply> answer =
+                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");
+
+        // ask()'s own unlocked timeout cleanup can forget this exact turnId at essentially the same
+        // instant answer() has already unblocked the worker and is now waiting on the resumed turn's
+        // real reply — reproduce that interleaving directly instead of trying to win a real race.
+        messages.forgetTurnForTest(turnId);
+
+        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
+                "the primary's own answer() call must still see the worker's real reply");
+        MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+        assertEquals("PR opened: https://example/pulls/42", done.reply(),
+                "fleet_poll{ticket} must return the worker's real reply, not stay PENDING forever "
+                        + "just because ask()'s timeout cleanup forgot this turnId first");
+    }
+
+    /**
+     * fleetd #329 (F3). {@link MessageService#reply} reads the same orphan {@code Task}'s {@code
+     * turnId} twice on this path (once to check it is non-null, once as the {@code
+     * asyncTasksByTurn.remove} key) — the exact double-read shape fleetd #324 fixed in {@code
+     * finishAsyncTask}. This test drives the same real sequence as {@code
+     * aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket} (worker asks, primary answers, the
+     * primary's own bounded wait for the resumed turn expires) to reach a task with {@code turnId}
+     * genuinely stamped and no live rendezvous waiter open — the state {@code askAnsweredAsyncTasks}
+     * matches here — then fires the identical production cleanup {@code ask()}'s own timeout path
+     * runs ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd
+     * #324 merged) at the point between the check and the removal use, via a dedicated test hook
+     * mirroring {@code finishAsyncTaskRaceHook}.
+     */
+    @Test
+    void replyToAnOrphanedTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() 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();
+
+        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 "
+                        + "live waiter — the state reply()'s F3 code path matches");
+
+        messages.setReplyOrphanTurnIdRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
+        try {
+            assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+            MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+            assertEquals("PR opened: https://example/pulls/42", done.reply(),
+                    "the ticket must still resolve to the worker's real reply despite the forced race");
+        } finally {
+            messages.setReplyOrphanTurnIdRaceHookForTest(null);
+        }
+    }
+
     /**
      * fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
      * becomes false the instant it lapses — proven above), so a second, independent delegation can
@@ -1103,6 +1255,63 @@ class MessageServiceTest {
         assertEquals("done", done.reply());
     }
 
+    // --- fleetd #329 (F2): an exception after the ticket's future completes must reach a log -----
+    //
+    // sendAsync's own executor task ends with catch (Throwable t) { task.future.completeExceptionally(t); }
+    // — but finishAsyncTask completes that same future on its first line, so anything that throws
+    // afterward hits an already-completed future: completeExceptionally returns false and does
+    // nothing, and (before this fix) nothing logged it either. Measured on the ticket: temporarily
+    // reintroducing the fleetd #324 NPE reproduced 19 real exceptions on the ordinary path with
+    // 74/74 tests staying green and zero log lines. There is no reachable production call site where
+    // finishAsyncTask throws after completing the future (fleetd #324 already closed the one that
+    // used to), so this test drives the shape directly via a dedicated test-only hook
+    // (afterFinishAsyncTaskCompleteHookForTest) rather than trying to engineer a real exception into
+    // that narrow window — the same technique fleetd #324's own race test and this ticket's F1/F3
+    // tests use for their own races.
+
+    private static ListAppender<ILoggingEvent> attachMessageServiceLog() {
+        Logger logger = (Logger) LoggerFactory.getLogger(MessageService.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        return appender;
+    }
+
+    private static void detachMessageServiceLog(ListAppender<ILoggingEvent> appender) {
+        ((Logger) LoggerFactory.getLogger(MessageService.class)).detachAppender(appender);
+    }
+
+    @Test
+    void anExceptionAfterTheTicketFutureCompletesStillReachesTheLog() throws Exception {
+        ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
+        try {
+            messages.setAfterFinishAsyncTaskCompleteHookForTest(() -> {
+                throw new RuntimeException("PROBE-329-F2");
+            });
+
+            String ticket = messages.sendAsync(T, "do the task");
+            awaitWaiting();
+            assertTrue(rendezvous.resolve(T, "done"));
+
+            // finishAsyncTask's own first line already completed the future with the real result
+            // BEFORE the hook threw — F2 is a visibility gap, not a correctness gap for the ticket
+            // itself, so the ticket's own outcome must be unaffected by the swallowed exception.
+            MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+            assertEquals("done", done.reply());
+
+            assertTrue(appender.list.stream().anyMatch(e ->
+                            e.getLevel() == Level.ERROR
+                                    && e.getFormattedMessage().contains(ticket)
+                                    && e.getThrowableProxy() != null
+                                    && "PROBE-329-F2".equals(e.getThrowableProxy().getMessage())),
+                    "an exception thrown after the ticket's future already completed must still "
+                            + "reach the log, not vanish silently");
+        } finally {
+            messages.setAfterFinishAsyncTaskCompleteHookForTest(null);
+            detachMessageServiceLog(appender);
+        }
+    }
+
     // --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
 
     @Test