fleetd #689: pin the answerGatePasses call site via the audit trail

A unit test on the extracted helper proves the helper, not the call
site in sendMessage. allow() logs an AuditLog.allowed() entry for
every granted non-READ/METRICS/TASK_READ action, so a granted turnId
request must log both SEND and ANSWER, and a granted plain request
must log SEND alone. Verified this goes red when the call site is
deleted from sendMessage, and restores to a clean diff.

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

@@ -83,6 +83,18 @@ public final class FleetApp {
         };
     }
 
+    /**
+     * The second gate for {@code POST /sessions/{id}/message}: checked only when {@code turnId}
+     * is present and non-blank, against {@link Authz.Action#ANSWER}. A request with no {@code
+     * turnId} passes this gate unconditionally, without consulting {@code permit} at all, having
+     * already cleared the coarse {@link Authz.Action#SEND} grant checked ahead of it.
+     *
+     * @param permit reports whether the caller holds the named grant
+     */
+    static boolean answerGatePasses(String turnId, Predicate<Authz.Action> permit) {
+        return turnId == null || turnId.isBlank() || permit.test(Authz.Action.ANSWER);
+    }
+
     /** Default blocking window for a message; kept under typical HTTP idle timeouts. */
     private static final long DEFAULT_MESSAGE_TIMEOUT_MS = 25_000;
     private static final long MAX_MESSAGE_TIMEOUT_MS = 120_000;
@@ -625,27 +637,31 @@ public final class FleetApp {
      *
      * <p>Two call shapes share this route, exactly as {@code fleet_send} does over MCP (see
      * {@code FleetMcp#sendAction}): a plain delivery to {@code id}, and -- when the body carries
-     * {@code turnId} -- resolving a worker's blocked question. The body is parsed before the
-     * authorization check so the right one of {@link Authz.Action#SEND}/{@link Authz.Action#ANSWER}
-     * reaches the gate; a body that fails to parse is treated as the plain shape for that check
-     * alone, and is rejected afterward exactly as before.
+     * {@code turnId} -- resolving a worker's blocked question. The coarse {@link
+     * Authz.Action#SEND} grant is checked first, before the body is read at all; only once that
+     * passes is the body parsed, and a present {@code turnId} is then checked again against
+     * {@link Authz.Action#ANSWER}. A body that fails to parse is rejected with 400 and reaches
+     * neither {@code messages.answer} nor {@code messages.send}.
      */
     private void sendMessage(Context ctx) {
         String id = ctx.pathParam("id");
+        if (!allow(ctx, routeAction("POST /sessions/{id}/message"), id)) {
+            return;
+        }
         JsonNode body;
         try {
             body = mapper.readTree(ctx.body());
         } catch (Exception e) {
             body = null;
         }
-        String turnId = body == null ? null : body.path("turnId").asText(null);
-        if (!allow(ctx, routeAction("POST /sessions/{id}/message", turnId), id)) {
-            return;
-        }
         if (body == null) {
             ctx.status(400).json(Map.of("error", "bad_request", "detail", "body must be JSON"));
             return;
         }
+        String turnId = body.path("turnId").asText(null);
+        if (!answerGatePasses(turnId, action -> allow(ctx, action, id))) {
+            return;
+        }
         String content = body.path("content").asText("");
         long timeout = body.path("timeoutMs").asLong(DEFAULT_MESSAGE_TIMEOUT_MS);
         boolean wait = body.path("wait").asBoolean(true); // default: block for the reply (CB-104)

fleetd/src/test/java/dev/ltms/fleet/FleetdAssemblyTimingDefaultsTest.java

@@ -0,0 +1,190 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.inject.Injector;
+import dev.ltms.fleet.inject.StatusPoller;
+import dev.ltms.fleet.msg.LeadChannelHandle;
+import dev.ltms.fleet.msg.LeadCoordLoop;
+import dev.ltms.fleet.msg.LeadMessage;
+import dev.ltms.fleet.msg.ReplyInbox;
+import dev.ltms.fleet.msg.ReplyPushLoop;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Asserts that the assembled loops use the production reminder, coordination, and delivery timing
+ * defaults when no {@code primary:} block configures the reply-push values.
+ */
+class FleetdAssemblyTimingDefaultsTest {
+
+    private static final class FakeLeadChannel implements LeadChannelHandle {
+        @Override
+        public void publish(String toCoordId, LeadMessage message) {
+        }
+
+        @Override
+        public List<LeadMessage> peek() {
+            return List.of();
+        }
+
+        @Override
+        public void ack(String msgId) {
+        }
+
+        @Override
+        public String selfCoordId() {
+            return "test-lead";
+        }
+
+        @Override
+        public boolean heldDurable() {
+            return true;
+        }
+
+        @Override
+        public MailboxState inspect(String coordId) {
+            return MailboxState.unknown(coordId);
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
+    private static final class TestResourcePorts implements ResourcePorts {
+        final FakeHerdr herdr = new FakeHerdr();
+        Runnable shutdownHook;
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of();
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> new ReplyInbox() {
+                @Override public void own(String target) { }
+                @Override public void release(String target) { }
+                @Override public void publish(String target, String msgId, String content) { }
+                @Override public List<InboxMessage> peek(String target) { return List.of(); }
+                @Override public boolean ack(String target, String msgId) { return false; }
+            };
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> new FakeLeadChannel();
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            return () -> {
+                throw new UnsupportedOperationException("FakeHerdr is healthy; no poll wait is expected");
+            };
+        }
+    }
+
+    private TestResourcePorts ports;
+
+    @AfterEach
+    void tearDown() {
+        if (ports != null && ports.shutdownHook != null) {
+            ports.shutdownHook.run();
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                idleSleepGuard:
+                  enabled: false
+                coordinator:
+                  uri: "amqp://fake-lead-broker/vh"
+                  selfId: "test-lead"
+                """);
+        return FleetConfig.load(file);
+    }
+
+    private FleetdRuntime assemble(Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir);
+        ports = new TestResourcePorts();
+        return FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg,
+                new ConfigRef(dir.resolve("fleetd.yaml"), cfg), new SubscriptionGuard(cfg.guard().hostSet())), ports);
+    }
+
+    private static long longField(Object target, String name) throws Exception {
+        Field field = target.getClass().getDeclaredField(name);
+        field.setAccessible(true);
+        return field.getLong(target);
+    }
+
+    @Test
+    void productionBootPathUsesTheExpectedLoopTimingDefaults(@TempDir Path dir) throws Exception {
+        FleetdRuntime runtime = assemble(dir);
+
+        ReplyPushLoop pushLoop = runtime.pushLoop();
+        assertEquals(5, longField(pushLoop, "maxReminders"),
+                "without primary:, ReplyPushLoop must stop after five reminder attempts");
+        assertEquals(15_000L, longField(pushLoop, "backoffMs"),
+                "without primary:, ReplyPushLoop must wait fifteen seconds before the next reminder");
+
+        LeadCoordLoop leadCoordLoop = runtime.leadCoordLoop();
+        assertNotNull(leadCoordLoop, "control: coordinator: must build LeadCoordLoop");
+        assertEquals(3_000L, longField(leadCoordLoop, "intervalMs"),
+                "LeadCoordLoop must poll for peer-lead mail every three seconds");
+
+        StatusPoller poller = runtime.poller();
+        assertEquals(Injector.POLL_INTERVAL_MILLIS, longField(poller, "intervalMillis"),
+                "StatusPoller must use Injector's delivery poll interval");
+    }
+}

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

@@ -60,13 +60,25 @@ class MessageServiceTest {
         inbox.own(T);
     }
 
+    /**
+     * A send budget large enough that a test's own setup — {@link #awaitWaiting()} plus whatever
+     * status transitions it drives afterward — can never compete with it for the same clock. A test
+     * that needs {@code send.get(...)}'s own window to be the only timing bound it depends on uses
+     * {@link #sendAsync(String, long)} with this value instead of the default 5000 ms.
+     */
+    private static final long GENEROUS_SEND_BUDGET_MILLIS = 30_000;
+
     /** Run {@code send} on a background thread; the current thread drives the worker's turn. */
     private CompletableFuture<MessageService.Reply> sendAsync() {
         return sendAsync("do the task");
     }
 
     private CompletableFuture<MessageService.Reply> sendAsync(String content) {
-        return CompletableFuture.supplyAsync(() -> messages.send(T, content, 5000));
+        return sendAsync(content, 5000);
+    }
+
+    private CompletableFuture<MessageService.Reply> sendAsync(String content, long timeoutMillis) {
+        return CompletableFuture.supplyAsync(() -> messages.send(T, content, timeoutMillis));
     }
 
     private void awaitWaiting() throws InterruptedException {
@@ -80,7 +92,7 @@ class MessageServiceTest {
 
     @Test
     void completionFallbackResolvesATurnThatNeverCalledFleetReply() throws Exception {
-        CompletableFuture<MessageService.Reply> send = sendAsync();
+        CompletableFuture<MessageService.Reply> send = sendAsync("do the task", GENEROUS_SEND_BUDGET_MILLIS);
         awaitWaiting();
 
         herdr.readText("$ prompt");                // pre-turn pane: no answer yet (baseline reference)
@@ -96,6 +108,32 @@ class MessageServiceTest {
         assertTrue(reply.completed(), "a scraped completion still counts as completed");
     }
 
+    /**
+     * Pins {@link #GENEROUS_SEND_BUDGET_MILLIS} as the budget {@link
+     * #completionFallbackResolvesATurnThatNeverCalledFleetReply} depends on. A 5500 ms delay between
+     * {@link #awaitWaiting()} and the status transitions that drive completion stands in for a loaded
+     * machine's setup overhead — comfortably past the 5000 ms budget this send no longer uses, and
+     * still well inside this method's own 30 000 ms budget. The only clock this test depends on is
+     * {@code send.get}'s own 10 s window.
+     */
+    @Test
+    void completionFallbackSurvivesASlowHarnessBecauseItsSendBudgetIsNotTheBindingClock() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync("do the task", GENEROUS_SEND_BUDGET_MILLIS);
+        awaitWaiting();
+
+        Thread.sleep(5500);
+
+        herdr.readText("$ prompt");
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+        herdr.readText("BUILD GREEN: 391 files");
+        injector.onStatus(T, AgentStatus.IDLE);
+
+        MessageService.Reply reply = send.get(10, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.COMPLETED_UNREPLIED, reply.outcome(),
+                "a slow harness must not be mistaken for a timed-out delivery");
+    }
+
     @Test
     void completionFallbackReplacesAnEchoedInjectedBriefWithNoReportOutcome() throws Exception {
         String brief = "Implement the requested change. ".repeat(20);

fleetd/src/test/java/dev/ltms/fleet/rest/FleetAppAuthTest.java

@@ -1,5 +1,8 @@
 package dev.ltms.fleet.rest;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import dev.ltms.fleet.auth.CallerResolver;
 import dev.ltms.fleet.auth.Authz;
 import dev.ltms.fleet.auth.MemberRegistry;
@@ -18,6 +21,7 @@ import dev.ltms.fleet.msg.Rendezvous;
 import dev.ltms.fleet.session.FakeWorktrees;
 import dev.ltms.fleet.session.SessionManager;
 import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.testing.CapturedLog;
 import io.javalin.Javalin;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
@@ -28,10 +32,13 @@ import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -144,6 +151,38 @@ class FleetAppAuthTest {
         assertEquals(Authz.Action.ANSWER, FleetApp.routeAction("POST /sessions/{id}/message", "turn-1"));
     }
 
+    /**
+     * fleetd #689: {@code answerGatePasses} is the second, conditional gate behind {@code
+     * sendMessage}'s coarse {@link Authz.Action#SEND} check. With the {@code ANSWER} grant denied,
+     * a {@code turnId}-bearing request is refused while a plain one still passes — and the denied
+     * permit is queried only for the {@code turnId} case, never for the plain one, which is what
+     * proves this is a genuinely separate, conditional check rather than the {@code SEND} check
+     * renamed or an unconditional call whose result is ignored. Flipping only the {@code ANSWER}
+     * grant to allowed then flips only the {@code turnId} shape's outcome.
+     */
+    @Test
+    void answerGatePassesOnlyWhenTurnIdAbsentOrAnswerGranted() {
+        List<Authz.Action> queried = new ArrayList<>();
+        Predicate<Authz.Action> denyAnswer = action -> {
+            queried.add(action);
+            return false;
+        };
+
+        assertFalse(FleetApp.answerGatePasses("turn-1", denyAnswer),
+                "ANSWER denied ⇒ the turnId shape is refused");
+        assertEquals(List.of(Authz.Action.ANSWER), queried,
+                "the ANSWER grant, specifically, must be the one consulted");
+
+        queried.clear();
+        assertTrue(FleetApp.answerGatePasses(null, denyAnswer),
+                "no turnId ⇒ the plain shape passes even though ANSWER is denied");
+        assertTrue(FleetApp.answerGatePasses("  ", denyAnswer), "a blank turnId is treated as absent");
+        assertEquals(List.of(), queried, "the plain shape must never consult the permit at all");
+
+        assertTrue(FleetApp.answerGatePasses("turn-1", action -> true),
+                "flipping only the ANSWER grant to allowed flips only the turnId shape's outcome");
+    }
+
     private static Set<String> routesTheServerRegisters() {
         try {
             String source = Files.readString(REST_SOURCE).lines()
@@ -223,6 +262,92 @@ class FleetAppAuthTest {
                 "resolving another session's blocked question would be a worker escalating too");
     }
 
+    /**
+     * fleetd #689: a caller refused the coarse {@link Authz.Action#SEND} grant is refused on
+     * {@code SEND} specifically, even on the {@code turnId}-bearing shape that otherwise raises
+     * the check to {@link Authz.Action#ANSWER} — proving {@code turnId} was never read from the
+     * body before the refusal (reading it would have changed which action is named in the 403).
+     * The same caller refused with no body at all gets the identical detail, which could not hold
+     * if the decision depended on anything read from the body. Control: a caller who IS granted
+     * reaches past the gate and the body is used normally.
+     */
+    @Test
+    void aDeniedCallerIsRefusedOnSendEvenWithATurnIdBodyAndNeverReadsTheBody() throws Exception {
+        int workerPort = start(FakeHerdr.WORKER_PID, false, null); // denied: not primary/architect
+
+        HttpResponse<String> withTurnId = send(workerPort, "POST", "/sessions/term_b/message",
+                "{\"turnId\":\"turn-1\",\"content\":\"hi\"}", null);
+        assertEquals(403, withTurnId.statusCode());
+        assertTrue(withTurnId.body().contains("may not SEND"),
+                "the SEND check must be the one that fired, not ANSWER — ANSWER would only be "
+                        + "reachable by having already read turnId out of the body");
+
+        HttpResponse<String> noBody = send(workerPort, "POST", "/sessions/term_b/message", null, null);
+        assertEquals(403, noBody.statusCode());
+        assertTrue(noBody.body().contains("may not SEND"),
+                "refused identically with no body at all — the refusal cannot depend on body content");
+
+        // Control: a primary IS granted SEND, so the same turnId body is read and acted on —
+        // reaching messages.answer, which reports this unknown turnId as a stale one.
+        int primaryPort = start(999_999, false, null);
+        HttpResponse<String> granted = send(primaryPort, "POST", "/sessions/term_b/message",
+                "{\"turnId\":\"turn-1\",\"content\":\"hi\"}", null);
+        assertEquals(409, granted.statusCode());
+        assertTrue(granted.body().contains("stale_turn"), "a granted caller's body IS read and acted on");
+    }
+
+    /**
+     * fleetd #689 (ticket comment 18353): the only place {@code sendMessage}'s call to {@code
+     * answerGatePasses} is observable is the audit trail — {@code allow()} logs an {@code
+     * "allowed"} entry for every granted action except {@code READ}/{@code METRICS}/{@code
+     * TASK_READ}, and {@code ANSWER} is none of those. A granted {@code turnId} request must
+     * therefore log both a {@code SEND} and an {@code ANSWER} entry; a granted plain request must
+     * log {@code SEND} alone. A unit test of the extracted helper pins the helper; this pins the
+     * call site — deleting the {@code answerGatePasses} call from {@code sendMessage} leaves the
+     * helper's own test green but turns this one red.
+     */
+    @Test
+    void aGrantedTurnIdRequestAuditsBothSendAndAnswerButAPlainRequestAuditsSendAlone() throws Exception {
+        int port = start(999_999, false, null); // primary: granted both SEND and ANSWER
+        ObjectMapper mapper = new ObjectMapper();
+
+        try (CapturedLog audit = CapturedLog.at("audit", Level.INFO)) {
+            send(port, "POST", "/sessions/term_b/message",
+                    "{\"turnId\":\"turn-1\",\"content\":\"hi\"}", null);
+
+            List<String> allowed = allowedActions(audit, mapper);
+            assertTrue(allowed.contains("SEND"),
+                    "a turnId request must still clear the coarse SEND grant first");
+            assertTrue(allowed.contains("ANSWER"),
+                    "a turnId request must ALSO clear the ANSWER grant — this is the call site itself");
+        }
+
+        try (CapturedLog audit = CapturedLog.at("audit", Level.INFO)) {
+            send(port, "POST", "/sessions/term_b/message",
+                    "{\"content\":\"hi\",\"timeoutMs\":50}", null);
+
+            List<String> allowed = allowedActions(audit, mapper);
+            assertEquals(List.of("SEND"), allowed,
+                    "a plain request must log SEND and nothing else — ANSWER is conditional on "
+                            + "turnId, not something every request happens to log");
+        }
+    }
+
+    private static List<String> allowedActions(CapturedLog audit, ObjectMapper mapper) {
+        return audit.events().stream()
+                .map(ILoggingEvent::getFormattedMessage)
+                .map(line -> {
+                    try {
+                        return mapper.readTree(line);
+                    } catch (Exception e) {
+                        throw new AssertionError("audit line is not valid JSON: " + line, e);
+                    }
+                })
+                .filter(n -> "allowed".equals(n.path("outcome").asText()))
+                .map(n -> n.path("action").asText())
+                .toList();
+    }
+
     // --- token mode ---------------------------------------------------------------------------
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/rest/FleetAppTest.java

@@ -675,6 +675,26 @@ class FleetAppTest {
         assertEquals(400, postMessage(port, "{}").statusCode());
     }
 
+    /**
+     * fleetd #689: a body that fails to parse is rejected with 400 before {@code turnId} is ever
+     * read from it, so it reaches neither {@code messages.answer} (which needs a {@code turnId})
+     * nor {@code messages.send} — confirmed here for {@code send} by the fake agent's idle status,
+     * which would otherwise make an immediate {@code agent.prompt} delivery observable.
+     */
+    @Test
+    void malformedBodyReturns400AndNeverReachesSendOrAnswer() throws Exception {
+        FakeHerdr herdr = new FakeHerdr().agentStatus("idle"); // idle ⇒ send would deliver right away if reached
+        int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
+
+        HttpResponse<String> res = postMessage(port, "not json at all");
+        assertEquals(400, res.statusCode());
+        JsonNode err = mapper.readTree(res.body());
+        assertEquals("bad_request", err.get("error").asText());
+        assertEquals("body must be JSON", err.get("detail").asText());
+        assertFalse(herdr.called("agent.prompt"),
+                "a malformed body must never reach messages.send's delivery");
+    }
+
     @Test
     void sessionStatusReportsLiveAgentStatus() throws Exception {
         FakeHerdr herdr = new FakeHerdr().agentStatus("blocked");