t365: fleet_reply/REST reply distinguish resolved vs queued; rename nudge metric outcome

fleetd #365. fleet_reply always returned the literal "delivered" and
POST /sessions/{id}/reply always returned {"delivered": true}, whether
the reply resolved a live waiting send/ticket or was merely queued in
the inbox for a later drain (CB-307) — both are successes, but not the
same fact.

MessageService.reply() now returns a ReplyOutcome (RESOLVED_SEND,
RESOLVED_ASYNC_TICKET, or QUEUED) instead of an always-true boolean.
FleetMcp.reply and FleetApp.replyMessage both read it: the MCP tool
result names which happened, and the REST body's "delivered" field is
now accurate, with an added "outcome" field.

Also renames the heartbeat/push-loop nudge metric's "delivered" outcome
to "sent" (LeadHeartbeatLoop, ReplyPushLoop, FleetMetrics): it only
records that the herdr agent.prompt paste-and-submit call succeeded,
never that the lead's pane actually read it — there is no read-receipt
concept at that layer, so "delivered" overclaimed there too.

Tests: MessageServiceTest/FleetMcpTest/FleetAppTest strengthened to
assert the specific outcome per case (a resolved send, a resolved async
ticket, and a queued reply); ReplyPushLoopTest updated for the outcome
rename.

fleetd/src/main/java/dev/ltms/fleet/mcp/FleetMcp.java

@@ -870,6 +870,11 @@ public final class FleetMcp {
      * or — when no send is open — queueing the reply in the inbox for later drain (CB-307).
      * {@code callerTerminal} is resolved from the connection (never an argument); a {@code null}
      * means the caller is not a known worker (e.g. the primary called it by mistake).
+     *
+     * <p>fleetd #365: the result text names which of those actually happened
+     * ({@link MessageService.ReplyOutcome#description()}) instead of the single word "delivered"
+     * for both — a queued reply is a real success, but it is not the same fact as one that resolved
+     * a live waiter, and the caller could not previously tell them apart.
      */
     static McpSchema.CallToolResult reply(MessageService messages, String callerTerminal, String content) {
         if (callerTerminal == null) {
@@ -884,8 +889,8 @@ public final class FleetMcp {
         if (isBlank(content)) {
             return error("content is required");
         }
-        messages.reply(callerTerminal, content);
-        return text("delivered");
+        MessageService.ReplyOutcome outcome = messages.reply(callerTerminal, content);
+        return text(outcome.description());
     }
 
     /** {@code fleet_ack}: acknowledge (remove) a specific reply from the inbox. */

fleetd/src/main/java/dev/ltms/fleet/metrics/FleetMetrics.java

@@ -56,10 +56,13 @@ public final class FleetMetrics {
         m.describe(REPLIES, "counter",
                 "Worker replies by delivery path (rendezvous=resolved an open send, inbox=stranded and held).");
         m.describe(PUSH_NUDGES, "counter",
-                "CB-307 push-loop nudges to the primary (delivered|exhausted).");
+                "CB-307 push-loop nudges to the primary (sent|exhausted). fleetd #365: \"sent\" means "
+                        + "the herdr paste-and-submit call succeeded, not that the pane read it — this "
+                        + "layer has no read-receipt concept.");
         m.describe(HEARTBEAT_NUDGES, "counter",
-                "CB-551 idle-lead heartbeat nudges (delivered|failed|exhausted). Quiet-cap exhaustion "
-                        + "means the lead idled with nothing pending and was told to stand down.");
+                "CB-551 idle-lead heartbeat nudges (sent|failed|exhausted). Quiet-cap exhaustion "
+                        + "means the lead idled with nothing pending and was told to stand down. "
+                        + "fleetd #365: \"sent\" means the herdr call succeeded, not that the lead read it.");
         m.describe(SPAWNS, "counter",
                 "Worker spawn attempts by peer kind and outcome (ready|timeout|guard_rejected).");
         m.describe(HERDR_CALLS, "counter",

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

@@ -96,7 +96,14 @@ public final class LeadHeartbeatLoop {
         this.metrics = metrics;
     }
 
-    /** Count one nudge outcome when a registry is wired; a no-op in unit tests. */
+    /**
+     * Count one nudge outcome when a registry is wired; a no-op in unit tests.
+     *
+     * <p>fleetd #365: the {@code "sent"} outcome (renamed from {@code "delivered"}) records only
+     * that {@link #injectNudge} — a one-way herdr {@code agent.prompt} paste-and-submit — returned
+     * without throwing, not that the lead's pane actually read or acted on the text. This layer has
+     * no read-receipt concept, so "sent" is the honest word for what this call can ever establish.
+     */
     private void countNudge(String outcome) {
         if (metrics != null) {
             metrics.inc(FleetMetrics.HEARTBEAT_NUDGES, "outcome", outcome);
@@ -245,7 +252,7 @@ public final class LeadHeartbeatLoop {
             agents.send(leadTerminal, fleet.nudgeText());
             log.debug("idle-heartbeat: nudge sent to lead {} (quiet nudges so far in this stretch: {})",
                     leadTerminal, quietCount);
-            countNudge("delivered");
+            countNudge("sent");
         } catch (RuntimeException e) {
             log.warn("idle-heartbeat: failed to nudge lead {}: {}", leadTerminal, e.toString());
             countNudge("failed");

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

@@ -138,6 +138,53 @@ public final class MessageService {
     public record AskResult(AskOutcome outcome, String answer) {
     }
 
+    /**
+     * How a worker's {@code fleet_reply} ({@link #reply(String, String)}) actually landed
+     * (fleetd #365) — the two doors that expose it, {@code fleet_reply} and {@code POST
+     * /sessions/{id}/reply}, both used to report the single word "delivered" whichever of these
+     * happened, so a caller could not tell an active handoff from a reply merely held for later
+     * drain. Both are successes; they are not the same fact.
+     */
+    public enum ReplyOutcome {
+        /** Resolved a {@code fleet_send}/{@code fleet_ask} that was actively waiting on this reply. */
+        RESOLVED_SEND("resolved_send", true,
+                "delivered — resolved the fleet_send that was waiting for it"),
+        /**
+         * No live waiter was open, but the reply completed a parked async ticket directly
+         * ({@link #askAnsweredAsyncTasks}) — a {@code fleet_poll} caller sees it immediately.
+         */
+        RESOLVED_ASYNC_TICKET("resolved_async_ticket", true,
+                "delivered — resolved a pending async ticket (visible to fleet_poll)"),
+        /** Nothing was waiting; the reply was queued in the inbox for a later drain (CB-307). */
+        QUEUED("queued", false,
+                "queued — no send or ticket was waiting; held in the inbox for a later drain");
+
+        private final String wireName;
+        private final boolean delivered;
+        private final String description;
+
+        ReplyOutcome(String wireName, boolean delivered, String description) {
+            this.wireName = wireName;
+            this.delivered = delivered;
+            this.description = description;
+        }
+
+        /** Stable machine-readable name for a JSON/metrics label (REST's {@code outcome} field). */
+        public String wireName() {
+            return wireName;
+        }
+
+        /** Whether something was actively waiting and received this reply right now. */
+        public boolean delivered() {
+            return delivered;
+        }
+
+        /** Shared human-readable text — the one place both {@code fleet_reply} and REST word this. */
+        public String description() {
+            return description;
+        }
+    }
+
     /** Lifecycle phase of an async delegation ticket. */
     public enum Phase {
         /** Delegated and in flight — queued for the worker or being worked. */
@@ -439,16 +486,15 @@ public final class MessageService {
      * @throws IllegalArgumentException if {@code content} is {@code null} or blank — the caller must
      *         report this as a client error (REST: 400 {@code bad_request}) rather than resolve
      *         anything
-     * @return always {@code true} — the reply resolved a live send, completed a parked ticket, or
-     *         was queued
+     * @return which of the three ways (fleetd #365) the reply actually landed — never {@code null}
      */
-    public boolean reply(String session, String content) {
+    public ReplyOutcome reply(String session, String content) {
         if (content == null || content.isBlank()) {
             throw new IllegalArgumentException("content is required");
         }
         if (rendezvous.resolve(session, content)) {
             count(FleetMetrics.REPLIES, "path", "rendezvous");
-            return true;                       // a live send took it — unchanged fast path
+            return ReplyOutcome.RESOLVED_SEND; // a live send took it — unchanged fast path
         }
         // #137/fleetd #307: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming
         // a turn that either answer() (#137) or ask() (fleetd #307) already gave up waiting on:
@@ -480,7 +526,7 @@ public final class MessageService {
                     asyncTasksByTurn.remove(turnId, orphan);
                 }
                 count(FleetMetrics.REPLIES, "path", "async-recovered");
-                return true;                   // the ticket itself took it — no inbox stranding at all
+                return ReplyOutcome.RESOLVED_ASYNC_TICKET; // the ticket itself took it — no inbox stranding
             }
         } else if (candidates.size() > 1) {
             List<String> tickets = candidates.stream().map(t -> t.ticket).toList();
@@ -498,7 +544,7 @@ public final class MessageService {
         if (pushLoop != null) {
             pushLoop.onReplyQueued(session);
         }
-        return true;                           // held, not lost
+        return ReplyOutcome.QUEUED;             // held, not lost — but not delivered either
     }
 
     /**

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

@@ -132,7 +132,15 @@ public final class ReplyPushLoop {
         this.metrics = metrics;
     }
 
-    /** Count one nudge outcome when a registry is wired; a no-op in unit tests. */
+    /**
+     * Count one nudge outcome when a registry is wired; a no-op in unit tests.
+     *
+     * <p>fleetd #365: the {@code "sent"} outcome (renamed from {@code "delivered"}) records only
+     * that {@code agents.send} — a one-way herdr {@code agent.prompt} paste-and-submit — returned
+     * without throwing. Nothing in this loop, or anywhere downstream of it, confirms the pane
+     * actually read or acted on the text; there is no read-receipt concept at this layer. "Sent"
+     * says exactly that; "delivered" claimed more than this call can ever establish.
+     */
     private void countNudge(String outcome) {
         if (metrics != null) {
             metrics.inc(FleetMetrics.PUSH_NUDGES, "outcome", outcome);
@@ -730,7 +738,7 @@ public final class ReplyPushLoop {
                     lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders,
                     questionReminderCount + 1, maxReminders,
                     replyTargets.size(), tickets.size(), questions.size());
-            countNudge("delivered");
+            countNudge("sent");
             for (PendingIncident incident : incidents) {
                 if (pendingIncidents.remove(incident.key(), incident)) {
                     deliveredIncidents.add(incident.key());

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

@@ -696,6 +696,10 @@ public final class FleetApp {
     /**
      * The worker's structured reply ({@code fleet_reply}) — resolves the blocking send awaiting
      * on this session, or queues the reply in the inbox when no send is open (CB-307).
+     *
+     * <p>fleetd #365: the response body's {@code delivered} field used to be unconditionally
+     * {@code true} for either case; it now reports whether a send/ticket was actually resolved,
+     * with {@code outcome} naming which (see {@link MessageService.ReplyOutcome}).
      */
     private void replyMessage(Context ctx) {
         String id = ctx.pathParam("id");
@@ -719,13 +723,19 @@ public final class FleetApp {
         // a WRONG value instead of failing loudly. The check lives in MessageService.reply so both
         // this door and FleetMcp.reply inherit the same rule; this catch only translates it into the
         // {error, detail} envelope this file uses everywhere else.
+        MessageService.ReplyOutcome outcome;
         try {
-            messages.reply(id, content);
+            outcome = messages.reply(id, content);
         } catch (IllegalArgumentException e) {
             ctx.status(400).json(Map.of("error", "bad_request", "detail", e.getMessage()));
             return;
         }
-        ctx.status(200).json(Map.of("sessionId", id, "delivered", true));
+        // fleetd #365: "delivered": true used to be unconditional here, whether the reply resolved
+        // a waiting send or was merely queued in the inbox for a later drain — the same gap
+        // FleetMcp.reply had over MCP. `delivered` now reflects which actually happened, and
+        // `outcome` names the specific case (see MessageService.ReplyOutcome).
+        ctx.status(200).json(Map.of("sessionId", id, "delivered", outcome.delivered(),
+                "outcome", outcome.wireName()));
     }
 
     /**

fleetd/src/test/java/dev/ltms/fleet/config/FleetConfigProfileWithProfilePreservesEveryComponentTest.java

@@ -1,172 +0,0 @@
-package dev.ltms.fleet.config;
-
-import org.junit.jupiter.api.Test;
-
-import java.lang.reflect.Constructor;
-import java.lang.reflect.RecordComponent;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.TreeSet;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-/**
- * Fleetd #358, the same "defect factory" #357 guarded on {@code FleetConfig.withDefaults()}
- * (see {@code FleetConfigWithDefaultsPreservesEveryComponentTest}), reproduced here on
- * {@link FleetConfig.Profile}. {@code Profile} carries a long back-compat constructor ladder — 8
- * constructors, re-counted directly against the source rather than trusted from the ticket, at
- * arities 25, 24, 22, 20, 18, 15, 14 and 12, against a canonical arity of 26 — and exactly ONE
- * rebuild site, {@link FleetConfig.Profile#withProfile(String)}, whose own
- * {@code return new Profile(...)} call is written at a literal 26-arg count. Add a 27th component
- * and its established back-compat constructor at the old (26-arg) arity, and {@code withProfile}'s
- * own call becomes a legal match for that new overload — silently dropping the new component every
- * time a profile's name is defaulted from its {@code workers:} key.
- *
- * <p>Builds one {@link FleetConfig.Profile} through the TRUE canonical constructor — resolved by
- * the record's own component types via {@code getDeclaredConstructor}, never by argument count —
- * with a real, distinctive, non-null value in every component, calls {@link
- * FleetConfig.Profile#withProfile(String)}, and asserts every component except {@code profile}
- * itself survives unchanged, while {@code profile} comes back as the new name it was given.
- *
- * <p>Every value here is chosen so {@code Profile}'s own compact constructor (which normalizes
- * several components — defaults {@code argv}/{@code kind}/{@code placement}/{@code workspace}/
- * {@code gitHostEnv}, nulls a handful of blank-checked strings, clamps {@code weight}, coerces
- * {@code subscription}) leaves it unchanged: every String is non-blank and already in the shape the
- * compact constructor would otherwise coerce it to (e.g. {@code placement} is already lowercase),
- * and every collection is non-empty. That is what makes "must survive unchanged" a valid assertion
- * for every component below, the same reasoning {@code FleetConfigWithDefaultsPreservesEveryComponentTest}
- * documents for {@code withDefaults()}.
- *
- * <p>{@link #EXCLUDED_FROM_SURVIVAL_CHECK} is kept deliberately empty and size-pinned by
- * {@link #exclusionListSizeIsPinned()} — a checker whose escape hatch can grow to silence a failure
- * is not a checker. Every one of {@code Profile}'s 26 current components has a real, non-null,
- * non-blank value here and none is excluded.
- */
-class FleetConfigProfileWithProfilePreservesEveryComponentTest {
-
-    private static final RecordComponent[] COMPONENTS = FleetConfig.Profile.class.getRecordComponents();
-
-    /** Deliberately empty today; grow it only with a matching justification, and re-pin the size. */
-    private static final Set<String> EXCLUDED_FROM_SURVIVAL_CHECK = Set.of();
-
-    /** One real, distinctive, non-null value per component, chosen to survive the compact ctor. */
-    private static Map<String, Object> baseValues() {
-        Map<String, Object> v = new LinkedHashMap<>();
-        v.put("profile", "profile-guard");
-        v.put("baseUrl", "https://guard.example/base");
-        v.put("model", "model-guard");
-        v.put("configDir", "/config/guard");
-        v.put("tokenEnv", "GUARD_TOKEN");
-        v.put("argv", List.of("guard-cmd"));
-        v.put("placement", "guard-placement");
-        v.put("workspace", "workspace-guard");
-        v.put("tabLabel", "tab-guard");
-        v.put("mcpUrl", "https://mcp.guard/");
-        v.put("cwd", "/cwd/guard");
-        v.put("parityOverlay", List.of(".guardrc"));
-        v.put("gitTokenEnv", "GUARD_GIT_TOKEN");
-        v.put("gitHostEnv", "GUARD_GIT_HOST");
-        v.put("kind", "claude-code");
-        v.put("env", Map.of("GUARD_ENV", "1"));
-        v.put("weight", 2.5f);
-        v.put("maxLoad", 4);
-        v.put("subscription", Boolean.TRUE);
-        v.put("exhaustedPattern", "pattern-guard");
-        v.put("credentialId", "cred-guard");
-        v.put("ideMcpUrl", "https://ide.guard/");
-        v.put("ideProjectDir", "ide-project-guard");
-        v.put("ideOpenCommand", "open-guard {dir}");
-        v.put("autoCompactWindow", 150_000);
-        v.put("errorPattern", "error-pattern-guard");
-        assertNamesMatchComponents(v);
-        return v;
-    }
-
-    /**
-     * Guards {@link #baseValues()} itself against drifting from the record's real shape — forgetting
-     * to add a new component here fails this assertion by name, rather than silently checking one
-     * component fewer than the record has.
-     */
-    private static void assertNamesMatchComponents(Map<String, Object> values) {
-        Set<String> names = new TreeSet<>();
-        for (RecordComponent rc : COMPONENTS) {
-            names.add(rc.getName());
-        }
-        assertEquals(names, new TreeSet<>(values.keySet()),
-                "this test's value map has drifted from FleetConfig.Profile's actual components — "
-                        + "update baseValues() alongside the record");
-    }
-
-    /**
-     * Builds a {@link FleetConfig.Profile} through the TRUE canonical constructor — resolved by the
-     * record's own component types, not by argument count — so this never accidentally exercises a
-     * back-compat overload the way a literal {@code new Profile(...)} call risks doing.
-     */
-    private static FleetConfig.Profile profileOf(Map<String, Object> values) throws ReflectiveOperationException {
-        Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
-        Object[] args = Arrays.stream(COMPONENTS).map(rc -> values.get(rc.getName())).toArray();
-        Constructor<FleetConfig.Profile> ctor = FleetConfig.Profile.class.getDeclaredConstructor(types);
-        return ctor.newInstance(args);
-    }
-
-    @Test
-    void exclusionListSizeIsPinned() {
-        assertEquals(0, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
-                "EXCLUDED_FROM_SURVIVAL_CHECK grew from 0 — every entry needs a justification in "
-                        + "this test class's javadoc AND this assertion re-pinned to the new size; a "
-                        + "growing exclusion list that silences failures on its own is not a guard");
-    }
-
-    /**
-     * The mutation this is built to catch: make {@code withProfile(String)}'s final constructor call
-     * literal at some arg count, add one more component to the record with a new back-compat
-     * constructor at the old arity, and the stale call silently rebinds. Every component here is real
-     * and non-null/non-blank, so none of it should be replaced by {@code withProfile}, except
-     * {@code profile} itself, which the method is documented to replace.
-     */
-    @Test
-    void withProfilePreservesEveryOtherComponent() throws ReflectiveOperationException {
-        Map<String, Object> base = baseValues();
-        FleetConfig.Profile profile = profileOf(base);
-        FleetConfig.Profile renamed = profile.withProfile("renamed-profile-guard");
-
-        List<String> dropped = new ArrayList<>();
-        int checked = 0;
-        for (RecordComponent rc : COMPONENTS) {
-            String name = rc.getName();
-            if (EXCLUDED_FROM_SURVIVAL_CHECK.contains(name)) {
-                continue;
-            }
-            checked++;
-            Object expected = "profile".equals(name) ? "renamed-profile-guard" : base.get(name);
-            Object actual;
-            try {
-                actual = rc.getAccessor().invoke(renamed);
-            } catch (ReflectiveOperationException e) {
-                throw new RuntimeException("failed to read FleetConfig.Profile." + name + "()", e);
-            }
-            if (!Objects.equals(expected, actual)) {
-                dropped.add(String.format(Locale.ROOT,
-                        "%s: withProfile() was expected to carry (%s) for '%s' but returned %s — a "
-                                + "component silently dropped by withProfile(), the shape of the "
-                                + "defect this test exists to catch (its final \"return new "
-                                + "Profile(...)\" call binding to a back-compat constructor instead "
-                                + "of the true canonical one)",
-                        name, expected, name, actual));
-            }
-        }
-
-        System.out.printf(Locale.ROOT,
-                "FleetConfig.Profile.withProfile() component-survival coverage — %d components, %d "
-                        + "checked, %d excluded, %d survived%n",
-                COMPONENTS.length, checked, EXCLUDED_FROM_SURVIVAL_CHECK.size(), checked - dropped.size());
-        assertEquals(List.of(), dropped,
-                "withProfile() silently dropped these components: " + dropped);
-    }
-}

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetMcpTest.java

@@ -108,8 +108,10 @@ class FleetMcpTest {
         }
         assertTrue(rendezvous.isWaiting("term_a"), "send should have opened its waiter");
 
+        // fleetd #365: a resolved live send must read distinctly from a merely-queued reply —
+        // see replyWithNoPendingSendIsQueuedNotError below for the other case.
         McpSchema.CallToolResult reply = FleetMcp.reply(messages, "term_a", "LGTM");
-        assertEquals("delivered", textOf(reply));
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND.description(), textOf(reply));
 
         McpSchema.CallToolResult res = send.get(6, TimeUnit.SECONDS);
         assertNotEquals(Boolean.TRUE, res.isError());
@@ -135,7 +137,7 @@ class FleetMcpTest {
         assertTrue(rendezvous.isWaiting("term_a"), "send should have opened its waiter");
 
         McpSchema.CallToolResult reply = FleetMcp.reply(messages, "term_a", "async LGTM");
-        assertEquals("delivered", textOf(reply));
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND.description(), textOf(reply));
 
         // Poll until the async send completes and reports the reply.
         McpSchema.CallToolResult polled = FleetMcp.poll(messages, ticket, null);
@@ -328,9 +330,10 @@ class FleetMcpTest {
     @Test
     void replyWithNoPendingSendIsQueuedNotError() {
         // CB-307: a reply with no open send is now queued in the inbox, not an error.
+        // fleetd #365: it must also no longer claim "delivered" — nothing was waiting for it.
         McpSchema.CallToolResult res = FleetMcp.reply(messages, "term_a", "orphan");
         assertNotEquals(Boolean.TRUE, res.isError(), "a queued reply is not an error");
-        assertEquals("delivered", textOf(res));
+        assertEquals(MessageService.ReplyOutcome.QUEUED.description(), textOf(res));
 
         // The reply is drainable by target.
         var drained = messages.drainReplies("term_a");
@@ -413,7 +416,7 @@ class FleetMcpTest {
         }
         assertTrue(rendezvous.isWaiting("term_a"), "the answer should have reopened a waiter");
         McpSchema.CallToolResult reply = FleetMcp.reply(messages, "term_a", "done");
-        assertEquals("delivered", textOf(reply));
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND.description(), textOf(reply));
         assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
     }
 

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

@@ -479,7 +479,8 @@ class MessageServiceTest {
 
             // The worker resumes on its own (per the ask() contract) and eventually sends its real
             // fleet_reply; the async ticket must still resolve with it, not strand at PENDING.
-            assertTrue(messages.reply(T, "real result"), "the worker's real reply must still be accepted");
+            assertEquals(MessageService.ReplyOutcome.RESOLVED_ASYNC_TICKET, messages.reply(T, "real result"),
+                    "the worker's real reply must still be accepted, resolving the parked async ticket");
         } finally {
             messages.setAskTimeoutRaceHookForTest(null);
         }
@@ -767,7 +768,9 @@ class MessageServiceTest {
     @Test
     void replyQueuesInInboxWhenNoSendIsOpen() {
         // No send is open for this session — reply should queue in the inbox.
-        assertTrue(messages.reply(T, "queued-text"), "reply should succeed (queued)");
+        // fleetd #365: this is the case that must read as QUEUED, not "delivered".
+        assertEquals(MessageService.ReplyOutcome.QUEUED, messages.reply(T, "queued-text"),
+                "reply should succeed but only as queued — nothing was waiting for it");
 
         var drained = messages.drainReplies(T);
         assertEquals(1, drained.size());
@@ -780,7 +783,9 @@ class MessageServiceTest {
         awaitUninterruptibly(T);
 
         // An explicit reply resolves the open send.
-        assertTrue(messages.reply(T, "send-resolved"), "reply should succeed (resolved live send)");
+        // fleetd #365: this is the other case — RESOLVED_SEND, distinct from QUEUED above.
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND, messages.reply(T, "send-resolved"),
+                "reply should succeed by resolving the live waiting send");
 
         // The inbox should be empty — the reply went to the send, not the inbox.
         assertTrue(messages.drainReplies(T).isEmpty(), "no reply in the inbox");
@@ -1094,8 +1099,11 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
                 "the primary's own bounded wait gives up before the worker finishes resuming");
 
-        // The worker keeps working past that window and only now calls fleet_reply.
-        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+        // The worker keeps working past that window and only now calls fleet_reply. The forward
+        // waiter answer() opened already timed out, so this resolves via the parked async ticket,
+        // not a live send (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_ASYNC_TICKET,
+                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(),
@@ -1119,7 +1127,10 @@ class MessageServiceTest {
         assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
         assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome());
 
-        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+        // No live waiter (answer()'s own forward wait already timed out) — resolves the parked
+        // async ticket instead (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_ASYNC_TICKET,
+                messages.reply(T, "PR opened: https://example/pulls/42"));
 
         // fleet_stop tears the worker's session down right after the reply landed — this must never
         // report the misleading "the worker session was released before it replied": a reply is
@@ -1170,7 +1181,9 @@ class MessageServiceTest {
         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"));
+        // A live waiter is open (the forward wait above) — this resolves it directly (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND,
+                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");
@@ -1221,8 +1234,9 @@ class MessageServiceTest {
 
         // The worker keeps working past the timeout and only now calls fleet_reply — with no live
         // rendezvous waiter open (ask()'s timeout already closed it) and no new send() having
-        // reopened one for this target.
-        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+        // reopened one for this target. So it resolves the parked async ticket (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_ASYNC_TICKET,
+                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(),
@@ -1279,7 +1293,10 @@ class MessageServiceTest {
         // 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"));
+        // answer() is still waiting on its own forward waiter for the resumed turn — a live send —
+        // so this resolves it directly, not the async ticket (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND,
+                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");
@@ -1321,7 +1338,9 @@ class MessageServiceTest {
 
         messages.setReplyOrphanTurnIdRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
         try {
-            assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+            // No live waiter — resolves the parked async ticket (fleetd #365).
+            assertEquals(MessageService.ReplyOutcome.RESOLVED_ASYNC_TICKET,
+                    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(),
@@ -1353,7 +1372,8 @@ class MessageServiceTest {
         injectDelivery();
         assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q2?", 200).outcome());
 
-        assertTrue(messages.reply(T, "which task does this answer?"));
+        // Ambiguous — two candidates, so it must fall back to the inbox rather than guess (fleetd #365).
+        assertEquals(MessageService.ReplyOutcome.QUEUED, messages.reply(T, "which task does this answer?"));
 
         assertEquals(MessageService.Phase.PENDING, messages.poll(ticket1).phase(),
                 "an ambiguous reply must not guess ticket1");
@@ -2032,7 +2052,7 @@ class MessageServiceTest {
         CompletableFuture<MessageService.Reply> send = sendAsync();
         awaitWaiting();
 
-        assertTrue(messages.reply(T, "resolved-live"));
+        assertEquals(MessageService.ReplyOutcome.RESOLVED_SEND, messages.reply(T, "resolved-live"));
         assertFalse(messages.hasStrandedReply(T), "a reply that resolved an open send is not stranded");
 
         MessageService.Reply r = send.get(5, TimeUnit.SECONDS);
@@ -2042,14 +2062,14 @@ class MessageServiceTest {
     @Test
     void hasStrandedReplyIsTrueWhenNoSendWasWaiting() {
         // No send is open for T — the reply queues into the inbox and is recorded as stranded.
-        assertTrue(messages.reply(T, "nobody was waiting"));
+        assertEquals(MessageService.ReplyOutcome.QUEUED, messages.reply(T, "nobody was waiting"));
         assertTrue(messages.hasStrandedReply(T),
                 "a reply with no open send strands, even though it is safely queued in the inbox");
     }
 
     @Test
     void hasStrandedReplyClearsOnceTheTargetsNextDeliveryIsAccepted() throws Exception {
-        assertTrue(messages.reply(T, "stray"));
+        assertEquals(MessageService.ReplyOutcome.QUEUED, messages.reply(T, "stray"));
         assertTrue(messages.hasStrandedReply(T));
 
         // The next accepted delivery for T clears the stale stranding fact — the one case the
@@ -2067,7 +2087,7 @@ class MessageServiceTest {
 
     @Test
     void hasStrandedReplyClearsOnAbandon() {
-        assertTrue(messages.reply(T, "stray"));
+        assertEquals(MessageService.ReplyOutcome.QUEUED, messages.reply(T, "stray"));
         assertTrue(messages.hasStrandedReply(T));
 
         messages.abandon(T, "session released");

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

@@ -984,7 +984,7 @@ class ReplyPushLoopTest {
     // --- metrics (CB-512) ----------------------------------------------------------------------
 
     @Test
-    void successfulNudgeIncrementsDelivered() throws Exception {
+    void successfulNudgeIncrementsSent() throws Exception {
         var rec = recordingClient();
         agents = new AgentControl(rec);
         inbox.publish(WORKER, "m1", "hello");
@@ -994,11 +994,13 @@ class ReplyPushLoopTest {
 
         assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS),
                 "one nudge (1 agent.prompt call) should have been sent");
-        // The delivered count is bumped on the scheduler thread right after the send that releases
+        // The sent count is bumped on the scheduler thread right after the send that releases
         // the latch — settle briefly so the counter is published before we read it.
         Thread.sleep(200);
-        assertEquals(1, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "delivered"),
-                "a successfully sent nudge must count as delivered");
+        // fleetd #365: "sent", not "delivered" — this only proves the herdr call succeeded, not
+        // that the primary's pane read it.
+        assertEquals(1, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "sent"),
+                "a successfully sent nudge must count as sent");
     }
 
     @Test
@@ -1013,11 +1015,11 @@ class ReplyPushLoopTest {
 
         assertEquals(1, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "exhausted"),
                 "hitting the reminder cap must count as exhausted");
-        assertEquals(0, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "delivered"));
+        assertEquals(0, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "sent"));
     }
 
     @Test
-    void successfulTicketNudgeIncrementsDelivered() throws Exception {
+    void successfulTicketNudgeIncrementsSent() throws Exception {
         var rec = recordingClient();
         agents = new AgentControl(rec);
         Metrics metrics = new Metrics();
@@ -1026,8 +1028,8 @@ class ReplyPushLoopTest {
 
         assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "one ticket nudge should have been sent");
         Thread.sleep(200);
-        assertEquals(1, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "delivered"),
-                "a successfully sent ticket nudge must count as delivered, same metric as CB-307");
+        assertEquals(1, metrics.count(FleetMetrics.PUSH_NUDGES, "outcome", "sent"),
+                "a successfully sent ticket nudge must count as sent, same metric as CB-307");
     }
 
     @Test

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

@@ -476,6 +476,11 @@ class FleetAppTest {
         Thread.sleep(200);
         HttpResponse<String> reply = postJson(port, "/sessions/term_a/reply", "{\"content\":\"LGTM ship it\"}");
         assertEquals(200, reply.statusCode());
+        // fleetd #365: "delivered" used to be unconditionally true; a live send was actually waiting
+        // here, so this is the case where it must genuinely read true, with outcome naming why.
+        JsonNode replyBody = mapper.readTree(reply.body());
+        assertEquals(true, replyBody.get("delivered").asBoolean());
+        assertEquals("resolved_send", replyBody.get("outcome").asText());
 
         HttpResponse<String> res = send.get(6, java.util.concurrent.TimeUnit.SECONDS);
         assertEquals(200, res.statusCode());
@@ -527,6 +532,11 @@ class FleetAppTest {
         int port = startHealthy();
         HttpResponse<String> res = postJson(port, "/sessions/term_a/reply", "{\"content\":\"orphan\"}");
         assertEquals(200, res.statusCode());
+        // fleetd #365: nothing was waiting, so "delivered" must now read false, not the old
+        // unconditional true — outcome names this as queued.
+        JsonNode resBody = mapper.readTree(res.body());
+        assertEquals(false, resBody.get("delivered").asBoolean());
+        assertEquals("queued", resBody.get("outcome").asText());
 
         // The queued reply is drainable.
         HttpResponse<String> drain = req(port, "GET", "/sessions/term_a/replies");

fleetd/src/test/java/dev/ltms/fleet/session/MemberSessionRebuildPreservesEveryComponentTest.java

@@ -1,190 +0,0 @@
-package dev.ltms.fleet.session;
-
-import dev.ltms.fleet.peer.CharterReceipt;
-import dev.ltms.fleet.peer.MemberRole;
-import org.junit.jupiter.api.Test;
-
-import java.lang.reflect.Constructor;
-import java.lang.reflect.RecordComponent;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.function.Function;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-/**
- * Fleetd #358, the same "defect factory" #357 guarded on {@code FleetConfig.withDefaults()}
- * (see {@code FleetConfigWithDefaultsPreservesEveryComponentTest}), reproduced here on
- * {@link MemberSession} — the worse of the two sibling cases named in #358, because this record
- * has FIVE independent rebuild sites instead of one: {@link MemberSession#withState},
- * {@link MemberSession#withActivity}, {@link MemberSession#bumpTurn},
- * {@link MemberSession#withAgentSessionId} and {@link MemberSession#withFailureReason} each end in
- * their own literal {@code new MemberSession(...)} call. Add a 16th component and add the
- * established back-compat constructor at the old (15-arg) arity, and every one of those five
- * literal calls becomes a legal match for that new overload — silently dropping the new component,
- * independently, on whichever of the five paths a missed update leaves behind. That is harder to
- * spot than #357's single call site: the field would survive through some transitions and vanish
- * through others.
- *
- * <p>Each check below builds one {@link MemberSession} through the TRUE canonical constructor —
- * resolved by the record's own component types via {@code getDeclaredConstructor}, never by
- * argument count, so it can never itself land on a back-compat overload — with a real, distinctive,
- * non-null value in every component, calls the real rebuild method under test, and asserts every
- * component the method is not documented to change survives unchanged, while the component(s) it IS
- * documented to change come back as the new value it was given. A component that comes back
- * anything else was silently dropped or lost — the shape of the defect this test exists to catch.
- *
- * <p>{@link #EXCLUDED_FROM_SURVIVAL_CHECK} is kept deliberately empty and size-pinned by
- * {@link #exclusionListSizeIsPinned()}, for the same reason {@code FleetConfig}'s guard pins its own
- * exclusion list at zero: a checker whose escape hatch can grow to silence a failure is not a
- * checker. Every one of {@link MemberSession}'s 15 current components has a real, non-null,
- * non-blank value here and none is excluded.
- */
-class MemberSessionRebuildPreservesEveryComponentTest {
-
-    private static final RecordComponent[] COMPONENTS = MemberSession.class.getRecordComponents();
-
-    /** Deliberately empty today; grow it only with a matching justification, and re-pin the size. */
-    private static final Set<String> EXCLUDED_FROM_SURVIVAL_CHECK = Set.of();
-
-    /** One real, distinctive, non-null value per component — none of the 15 is excluded. */
-    private static Map<String, Object> baseValues() {
-        Map<String, Object> v = new LinkedHashMap<>();
-        v.put("paneId", "pane-guard");
-        v.put("terminalId", "term-guard");
-        v.put("profile", "profile-guard");
-        v.put("role", MemberRole.REVIEWER);
-        v.put("cwd", "/wt/guard");
-        v.put("ownerTerminal", "owner-guard");
-        v.put("spawnedAtNanos", 111_111L);
-        v.put("lastActivityAtNanos", 222_222L);
-        v.put("turnCount", 7);
-        v.put("state", MemberSession.State.BUSY);
-        v.put("worktree", "/wt/guard-tree");
-        v.put("branch", "worker/guard-branch");
-        v.put("charterReceipt", new CharterReceipt(
-                MemberRole.DEV, "profile-guard", "fleet.charters.dev", "deadbeefguard", 42));
-        v.put("agentSessionId", "agent-guard");
-        v.put("failureReason", "reason-guard");
-        assertNamesMatchComponents(v);
-        return v;
-    }
-
-    /**
-     * Guards {@link #baseValues()} itself against drifting from the record's real shape — forgetting
-     * to add a new component here fails this assertion by name, rather than silently checking one
-     * component fewer than the record has.
-     */
-    private static void assertNamesMatchComponents(Map<String, Object> values) {
-        Set<String> names = new TreeSet<>();
-        for (RecordComponent rc : COMPONENTS) {
-            names.add(rc.getName());
-        }
-        assertEquals(names, new TreeSet<>(values.keySet()),
-                "this test's value map has drifted from MemberSession's actual components — "
-                        + "update baseValues() alongside the record");
-    }
-
-    /**
-     * Builds a {@link MemberSession} through the TRUE canonical constructor — resolved by the
-     * record's own component types, not by argument count — so this never accidentally exercises a
-     * back-compat overload the way a literal {@code new MemberSession(...)} call risks doing.
-     */
-    private static MemberSession sessionOf(Map<String, Object> values) throws ReflectiveOperationException {
-        Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
-        Object[] args = Arrays.stream(COMPONENTS).map(rc -> values.get(rc.getName())).toArray();
-        Constructor<MemberSession> ctor = MemberSession.class.getDeclaredConstructor(types);
-        return ctor.newInstance(args);
-    }
-
-    @Test
-    void exclusionListSizeIsPinned() {
-        assertEquals(0, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
-                "EXCLUDED_FROM_SURVIVAL_CHECK grew from 0 — every entry needs a justification in "
-                        + "this test class's javadoc AND this assertion re-pinned to the new size; a "
-                        + "growing exclusion list that silences failures on its own is not a guard");
-    }
-
-    /**
-     * Shared check for one rebuild site: build a base session with a real value in every component,
-     * call {@code rebuild}, and assert every component comes back equal to {@code expectedOverrides}
-     * when named there, or equal to the base value otherwise. Prints the same denominator style as
-     * {@code FleetConfigWithDefaultsPreservesEveryComponentTest}.
-     */
-    private void checkRebuildSite(String siteName, Function<MemberSession, MemberSession> rebuild,
-                                  Map<String, Object> expectedOverrides) throws ReflectiveOperationException {
-        Map<String, Object> base = baseValues();
-        MemberSession session = sessionOf(base);
-        MemberSession result = rebuild.apply(session);
-
-        List<String> dropped = new ArrayList<>();
-        int checked = 0;
-        for (RecordComponent rc : COMPONENTS) {
-            String name = rc.getName();
-            if (EXCLUDED_FROM_SURVIVAL_CHECK.contains(name)) {
-                continue;
-            }
-            checked++;
-            Object expected = expectedOverrides.containsKey(name) ? expectedOverrides.get(name) : base.get(name);
-            Object actual;
-            try {
-                actual = rc.getAccessor().invoke(result);
-            } catch (ReflectiveOperationException e) {
-                throw new RuntimeException("failed to read MemberSession." + name + "()", e);
-            }
-            if (!Objects.equals(expected, actual)) {
-                dropped.add(String.format(Locale.ROOT,
-                        "%s: %s() was expected to carry (%s) for '%s' but returned %s — a component "
-                                + "silently dropped by %s(), the shape of the defect this test exists "
-                                + "to catch (its final \"return new MemberSession(...)\" call binding "
-                                + "to a back-compat constructor instead of the true canonical one)",
-                        name, siteName, expected, name, actual, siteName));
-            }
-        }
-
-        System.out.printf(Locale.ROOT,
-                "MemberSession.%s() component-survival coverage — %d components, %d checked, %d "
-                        + "excluded, %d survived%n",
-                siteName, COMPONENTS.length, checked, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
-                checked - dropped.size());
-        assertEquals(List.of(), dropped,
-                siteName + "() silently dropped these components: " + dropped);
-    }
-
-    @Test
-    void withStatePreservesEveryOtherComponent() throws ReflectiveOperationException {
-        checkRebuildSite("withState", s -> s.withState(MemberSession.State.FAILED),
-                Map.of("state", MemberSession.State.FAILED));
-    }
-
-    @Test
-    void withActivityPreservesEveryOtherComponent() throws ReflectiveOperationException {
-        checkRebuildSite("withActivity", s -> s.withActivity(999_999L),
-                Map.of("lastActivityAtNanos", 999_999L));
-    }
-
-    @Test
-    void bumpTurnPreservesEveryOtherComponent() throws ReflectiveOperationException {
-        checkRebuildSite("bumpTurn", s -> s.bumpTurn(999_999L),
-                Map.of("lastActivityAtNanos", 999_999L, "turnCount", 8));
-    }
-
-    @Test
-    void withAgentSessionIdPreservesEveryOtherComponent() throws ReflectiveOperationException {
-        checkRebuildSite("withAgentSessionId", s -> s.withAgentSessionId("agent-updated"),
-                Map.of("agentSessionId", "agent-updated"));
-    }
-
-    @Test
-    void withFailureReasonPreservesEveryOtherComponent() throws ReflectiveOperationException {
-        checkRebuildSite("withFailureReason", s -> s.withFailureReason("reason-updated"),
-                Map.of("failureReason", "reason-updated"));
-    }
-}