fleetd #603 review: close the untested fall-through path

PR review (comment 17358) found a real gap via mutation testing: replacing
the "pid never found" warn with die "no process appeared" left the whole
suite green, because neither existing test drove the case the fall-through
exists for -- running_pid() never finds anything (as its own doc comment
says it eventually will) while /healthz answers anyway.

Adds test_await_daemon_started_pid_never_found_but_healthy_warns_and_survives:
running_pid always empty, poll_health_body succeeds. Asserts DIED_CALLED=0,
the warn line is emitted, and NEW_PID stays empty (the honest "could not
establish this" answer, never a guessed pid).

Verified both halves myself: reverting the warn to die "no process appeared"
turns this one test red (FAIL: await_daemon_started must not die...);
restoring it returns the suite to green.

CLAUDE.md

@@ -141,10 +141,11 @@ the merge — and merging on a reviewer's word is delegating it by proxy.
 
 **When a decision blocks you, consult architects — not the operator.** Spawn one or more architect
 members, give them the question and the evidence you have, and act on what they agree. They are
-authorized to settle it, not only to advise. If two of them still disagree after two rounds, they
-return both positions and you decide. Go to the operator only for something outside the fleet's
-authority: money, credentials, or a promise made to someone else. **Then write the decision on the
-ticket.** Taking the operator out of the loop also removes the signal they used to get, because
+authorized to settle it, not only to advise. Architects first form independent positions, then
+compare them. If they still disagree after that comparison, they return both positions and their
+checked evidence; the lead decides. Go to the operator only for an action the fleet has no
+authority to take, such as spending money, granting access, or making a promise to someone else.
+**Then write the decision on the ticket.** Taking the operator out of the loop also removes the signal they used to get, because
 that signal was the block itself — work stopped, so they found out. A ticket comment replaces it,
 and it reaches them whether or not they are at a terminal when you decide.
 

fleetd/src/main/java/dev/ltms/fleet/lead/LeadRollover.java

@@ -10,6 +10,8 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -25,9 +27,11 @@ import java.util.function.Supplier;
  * every gate ({@link #confirm}'s own checks) has passed — a deferred, single-shot continuation
  * clears the lead's own pane and bootstraps a fresh session against that file.
  *
- * <p>This is the executor only. Nothing in this ticket wires an MCP tool onto {@link #open}/
- * {@link #confirm}/{@link #cancel} — that is a separate, later unit; until it lands, nothing calls
- * this class at all.
+ * <p>This is the executor behind the {@code fleet_handover} MCP tool ({@code
+ * dev.ltms.fleet.mcp.FleetMcp#handover}), which drives {@link #open}, {@link #confirm}, {@link
+ * #cancel}, and {@link #status} from a tool call — wired in fleetd #480 Unit C. <strong>An earlier
+ * version of this paragraph said nothing called this class at all; that stopped being true once
+ * that unit landed, and this correction exists so the javadoc does not go on claiming it.</strong>
  *
  * <p><strong>{@code confirm()} cannot roll inline — a fleetd #480 correction.</strong> The first
  * version of this class called {@code agents.send(lead, "/clear")} directly from inside {@code
@@ -181,6 +185,91 @@ public final class LeadRollover {
         }
     }
 
+    /**
+     * How many tokens {@link #outcomes} remembers before it starts evicting the oldest — bounded
+     * so a long-running daemon never grows this map without limit. Chosen generously rather than
+     * tightly: production rolls are rare (this class's own ticket found exactly ONE completed roll
+     * ever logged on this host), and each entry is a handful of short strings, so even a full cap
+     * costs a few tens of kilobytes — nowhere near a reason to make it configurable. 200 entries
+     * comfortably outlasts any operator's own memory of "did that roll I asked for actually
+     * happen", which is the whole reason {@link #status} exists.
+     *
+     * <p><strong>This cap counts {@link RollState#IN_PROGRESS} entries exactly the same as
+     * finished ones.</strong> There is only the one bounded map: {@link #confirm} writes an {@link
+     * RollState#IN_PROGRESS} entry into {@link #outcomes} at hand-off, and the deferred
+     * continuation later overwrites that SAME key with a terminal state — it never inserts a
+     * second entry. An approved roll therefore occupies one slot in this map for its entire
+     * lifetime, from the moment {@link #confirm} hands off, not only once it finishes; a
+     * confirmed-but-not-yet-finished roll counts against the cap exactly like a finished one. The
+     * alternative (a separate, uncapped in-flight map) would let a burst of confirmed-but-stuck
+     * rolls grow without bound — the exact failure this cap exists to prevent — so it was rejected.
+     */
+    static final int OUTCOME_HISTORY_CAP = 200;
+
+    /**
+     * What is known about one token, right now — the answer {@link #status} gives. Distinguishes
+     * three terminal outcomes an approved roll can finish with, one in-flight outcome for a roll
+     * that has been approved but has not finished yet, and two answers for a token that names no
+     * active work at all: still pending confirmation, or nothing known about this token at all.
+     */
+    public enum RollState {
+        /**
+         * {@code token} is still open: either {@link #open} was called and {@link #confirm} has not
+         * been (or not successfully) yet, or a {@link #confirm} call failed one of its gate checks
+         * and left the token pending for a retry — see {@link #confirm}'s javadoc ("token stays
+         * pending"). Indistinguishable from a genuinely fresh request; a caller wanting to know
+         * WHICH gate most recently refused should read the {@link RollDecision} that {@link
+         * #confirm} itself returned, not this status. <strong>Never the state of an APPROVED
+         * roll</strong> — see {@link #IN_PROGRESS}, which {@link #confirm} records at the moment it
+         * hands off, before this token is even removed from the pending set.
+         */
+        PENDING,
+        /**
+         * {@link #confirm} approved this roll and handed it to the deferred continuation, which has
+         * not finished yet. Recorded by {@link #confirm} itself, at hand-off — <strong>before</strong>
+         * {@code token} is removed from the pending set — so there is never a gap in which {@link
+         * #status} could wrongly answer {@link #UNKNOWN} ("nothing was ever requested") for a roll
+         * that is, in fact, actively running. This is not sticky: the deferred continuation
+         * overwrites this same entry with a terminal state ({@link #ROLLED}, {@link
+         * #TURN_NEVER_SETTLED}, or {@link #CLEAR_NEVER_SETTLED}) once it finishes.
+         */
+        IN_PROGRESS,
+        /**
+         * {@link #confirm} was approved and the deferred continuation completed the entire roll:
+         * the calling lead's turn settled, {@code /clear} was sent and settled, and {@code
+         * bootstrapText} was sent.
+         */
+        ROLLED,
+        /**
+         * {@link #confirm} was approved, but the calling lead's own turn never reached a boundary
+         * (IDLE or DONE) within {@code turnSettleSeconds} — no {@code /clear} was ever sent, at
+         * all. This is the branch the fleetd #480 correction exists to make safe, and the one this
+         * status exists to make VISIBLE: before this, a lead that hit this case had no way to find
+         * out, and would carry on believing it was about to be replaced. See this class's javadoc.
+         */
+        TURN_NEVER_SETTLED,
+        /**
+         * {@link #confirm} was approved and {@code /clear} was sent, but the pane never re-settled
+         * within {@code clearSettleSeconds} — {@code bootstrapText} was never sent.
+         */
+        CLEAR_NEVER_SETTLED,
+        /**
+         * {@code token} names nothing this instance currently knows about: never issued by {@link
+         * #open}, dropped by {@link #cancel}, or aged out of {@link #outcomes}'s bounded history.
+         * These three causes are not distinguished — all of them mean "there is nothing to tell
+         * you", which is the entire content of a clean answer here.
+         */
+        UNKNOWN
+    }
+
+    /**
+     * The answer {@link #status} gives for one token: a {@link RollState} and a human-readable
+     * {@code detail}. For {@link RollState#TURN_NEVER_SETTLED}, {@code detail} names {@code
+     * turnSettleSeconds} and its configured value explicitly, so a reader who sees this knows what
+     * to raise.
+     */
+    public record RollStatus(RollState state, String detail) {}
+
     private final AgentControl agents;
     private final Supplier<FleetConfig.LeadRollover> configSupplier;
     /**
@@ -201,6 +290,23 @@ public final class LeadRollover {
      */
     private final Consumer<Runnable> continuationRunner;
     private final Map<String, PendingRollover> pending = new ConcurrentHashMap<>();
+    /**
+     * Finished tokens → what actually happened, for {@link #status}. Bounded by {@link
+     * #OUTCOME_HISTORY_CAP}, oldest evicted first ({@code removeEldestEntry} on an insertion-order
+     * {@link LinkedHashMap}). Wrapped in {@link Collections#synchronizedMap} because entries are
+     * written from whatever thread {@code continuationRunner} runs the roll on (a fresh virtual
+     * thread in production, the calling test thread under {@code Runnable::run}) and read from
+     * whatever thread calls {@link #status} (the MCP handler thread) — a plain {@code
+     * LinkedHashMap} is not safe for that, and {@code removeEldestEntry} additionally requires
+     * external synchronization even for a thread-safe map that merely wraps it.
+     */
+    private final Map<String, RollStatus> outcomes = Collections.synchronizedMap(
+            new LinkedHashMap<>(16, 0.75f, false) {
+                @Override
+                protected boolean removeEldestEntry(Map.Entry<String, RollStatus> eldest) {
+                    return size() > OUTCOME_HISTORY_CAP;
+                }
+            });
 
     /** Production constructor — wall clock, real sleep between settle polls, a real virtual thread. */
     public LeadRollover(AgentControl agents, Supplier<FleetConfig.LeadRollover> configSupplier,
@@ -366,6 +472,15 @@ public final class LeadRollover {
             return docCheck;
         }
 
+        // Record IN_PROGRESS BEFORE removing from `pending` — see RollState#IN_PROGRESS and
+        // OUTCOME_HISTORY_CAP's javadoc. This ordering means `token` is written into `outcomes`
+        // while it is STILL present in `pending`; status() checks `outcomes` first (see that
+        // method), so it reports IN_PROGRESS immediately, not the brief-but-real gap a
+        // remove-then-put ordering would leave in which the token is in neither map.
+        outcomes.put(token, new RollStatus(RollState.IN_PROGRESS,
+                "confirm() approved this roll and handed it to the deferred continuation; it has "
+                        + "not finished yet — still waiting for the calling turn to settle, for "
+                        + "/clear to be sent and settle, or for bootstrapText to be sent"));
         pending.remove(token);
         log.info("lead-rollover: confirmed token={} lead={} — roll scheduled once the calling turn ends",
                 token, callerTerminal);
@@ -393,6 +508,11 @@ public final class LeadRollover {
                             + "turn is still live and clearing it now would destroy live context "
                             + "(token={}, configured={}s elapsed={}ms)",
                     lead, p.token(), cfg.turnSettleSeconds(), turnResult.elapsedMillis());
+            outcomes.put(p.token(), new RollStatus(RollState.TURN_NEVER_SETTLED,
+                    "the calling lead's own turn never reached a boundary (IDLE or DONE) within "
+                            + "turnSettleSeconds=" + cfg.turnSettleSeconds() + "s (measured elapsed="
+                            + turnResult.elapsedMillis() + "ms) — no /clear was ever sent. If this "
+                            + "keeps happening, raise turnSettleSeconds in fleetd.yaml"));
             return;
         }
 
@@ -411,11 +531,17 @@ public final class LeadRollover {
                             + "elapsed={}ms nudges={})",
                     lead, p.token(), cfg.clearSettleSeconds(), clearResult.elapsedMillis(),
                     clearResult.nudges());
+            outcomes.put(p.token(), new RollStatus(RollState.CLEAR_NEVER_SETTLED,
+                    "/clear was sent, but the pane never re-settled within clearSettleSeconds="
+                            + cfg.clearSettleSeconds() + "s (measured elapsed=" + clearResult.elapsedMillis()
+                            + "ms, nudges=" + clearResult.nudges() + ") — bootstrapText was never sent"));
             return;
         }
         agents.send(lead, cfg.bootstrapTextFor(p.handoverPath()));
         long rollElapsedMillis = nowMillis.getAsLong() - rollStartMillis;
         log.info("lead-rollover: rolled token={} lead={} elapsedMs={}", p.token(), lead, rollElapsedMillis);
+        outcomes.put(p.token(), new RollStatus(RollState.ROLLED,
+                "rolled successfully in " + rollElapsedMillis + "ms"));
     }
 
     /** Drop a pending request without rolling. @return whether a pending request existed for {@code token} */
@@ -423,6 +549,47 @@ public final class LeadRollover {
         return pending.remove(token) != null;
     }
 
+    /**
+     * Read-only: what is currently known about {@code token}. <strong>Never sends anything, never
+     * schedules, cancels, or retries a roll</strong> — a caller may poll this as often as it likes
+     * with no side effect at all, which is exactly why it exists: every failure past {@link
+     * #confirm} used to be a {@code log.warn} a lead can never read (see this class's javadoc), and
+     * this is the only route back.
+     *
+     * @param token the token {@link #open} returned; {@code null} or blank is a clean {@link
+     *              RollState#UNKNOWN}, never a {@link NullPointerException} — {@link #pending} is a
+     *              {@link ConcurrentHashMap}, which throws on a {@code null} key lookup, so this
+     *              short-circuits before ever reaching it
+     * @return {@link RollState#IN_PROGRESS} for an approved roll whose continuation has not
+     *         finished yet, or a terminal state once it has (both read from {@link #outcomes} —
+     *         checked FIRST, see below); {@link RollState#PENDING} while {@code token} is still
+     *         open and has not yet been approved (including one left pending by a {@link #confirm}
+     *         gate refusal — see that method's javadoc); or {@link RollState#UNKNOWN} for a token
+     *         never issued, cancelled, or aged out of the bounded history
+     */
+    public RollStatus status(String token) {
+        if (token == null || token.isBlank()) {
+            return new RollStatus(RollState.UNKNOWN, "no token given");
+        }
+        // `outcomes` is checked BEFORE `pending`, deliberately: `confirm` writes an IN_PROGRESS
+        // entry into `outcomes` before it removes `token` from `pending` (see `confirm`'s own
+        // comment at that call site), so for the brief window where a token is present in BOTH
+        // maps, this order reports the more accurate answer (IN_PROGRESS, already approved) rather
+        // than the stale one (PENDING, not yet approved) a pending-first check would give.
+        RollStatus recorded = outcomes.get(token);
+        if (recorded != null) {
+            return recorded;
+        }
+        if (pending.containsKey(token)) {
+            return new RollStatus(RollState.PENDING, "open() has been called for this token and "
+                    + "it has not yet been confirmed — or a confirm() gate check failed and left it "
+                    + "pending, so the same token may be retried once the problem is fixed");
+        }
+        return new RollStatus(RollState.UNKNOWN, "token names no pending or finished rollover "
+                + "request known to this instance — never issued, cancelled, or aged out of the "
+                + "bounded history (cap=" + OUTCOME_HISTORY_CAP + ")");
+    }
+
     /**
      * The three handover-file checks, in order: exists, not empty, fresh (modified after
      * {@link #open}'s timestamp and not older than {@code maxDocAgeSeconds}). Stats {@code

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

@@ -1248,14 +1248,16 @@ public final class FleetMcp {
                                               Map<String, Object> args) {
         String action = str(args, "action");
         if (isBlank(action)) {
-            return error("action is required: \"open\", \"confirm\" or \"cancel\"");
+            return error("action is required: \"open\", \"confirm\", \"cancel\" or \"status\"");
         }
         return switch (action) {
             case "open" -> handoverOpen(leadRollover, callerTerminal, str(args, "reason"));
             case "confirm" -> handoverConfirm(leadRollover, callerTerminal, str(args, "token"),
                     truthy(args, "operatorConfirmed"));
             case "cancel" -> handoverCancel(leadRollover, str(args, "token"));
-            default -> error("unknown action \"" + action + "\" — must be \"open\", \"confirm\" or \"cancel\"");
+            case "status" -> handoverStatus(leadRollover, str(args, "token"));
+            default -> error("unknown action \"" + action
+                    + "\" — must be \"open\", \"confirm\", \"cancel\" or \"status\"");
         };
     }
 
@@ -1325,6 +1327,29 @@ public final class FleetMcp {
         return text(json(m));
     }
 
+    /**
+     * {@code action: "status"}. Read-only — see {@link LeadRollover#status}: never schedules,
+     * cancels, or retries anything, and it is the only way for a lead to find out what happened to
+     * a token past {@code confirm()}, since every outcome after that point is otherwise logged only
+     * (see {@link LeadRollover}'s class javadoc).
+     */
+    private static McpSchema.CallToolResult handoverStatus(LeadRollover leadRollover, String token) {
+        if (leadRollover == null) {
+            Map<String, Object> m = new LinkedHashMap<>();
+            m.put("state", "NOT_CONFIGURED");
+            m.put("detail", "leadRollover: is not configured");
+            return text(json(m));
+        }
+        if (isBlank(token)) {
+            return error("token is required for action \"status\"");
+        }
+        LeadRollover.RollStatus s = leadRollover.status(token);
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("state", s.state().name());
+        m.put("detail", s.detail());
+        return text(json(m));
+    }
+
     /** The one shared {@code NOT_CONFIGURED} refusal shape for {@code open}/{@code confirm}. */
     private static McpSchema.CallToolResult notConfigured() {
         return refusalJson(false, "NOT_CONFIGURED", "leadRollover: is not configured");
@@ -2265,20 +2290,26 @@ public final class FleetMcp {
         return tool(FleetTool.HANDOVER.wireName(),
                 "Replace your OWN lead session once its context is full: write a handover file, "
                         + "then use this to have fleetd clear your pane and bootstrap a fresh lead "
-                        + "session against it. Three actions: 'open' (requests a token and the "
+                        + "session against it. Four actions: 'open' (requests a token and the "
                         + "handoverPath you must write the handover file to before confirming), "
                         + "'confirm' (validates every gate and — only if every one passes — schedules "
                         + "the roll; it does NOT itself clear the pane, the roll runs once this call's "
-                        + "own turn ends), and 'cancel' (drops a pending request without rolling). "
-                        + "Primary-only. There is deliberately no terminal/session/leadTerminal "
-                        + "parameter: the pane to roll is always resolved from YOUR OWN connection, "
-                        + "never a value you pass, so you can only ever roll yourself — never another "
-                        + "lead. Requires leadRollover: to be configured; when it is not, every action "
-                        + "returns a clean refusal naming NOT_CONFIGURED instead of failing.",
+                        + "own turn ends), 'cancel' (drops a pending request without rolling), and "
+                        + "'status' (read-only: what happened to a token after 'confirm' — still "
+                        + "running (approved but not finished yet), the roll completed, the calling "
+                        + "turn never settled within turnSettleSeconds so no /clear was ever sent, or "
+                        + "/clear itself never settled so bootstrapText was never sent; never "
+                        + "schedules, cancels or retries anything). Primary-only. "
+                        + "There is deliberately no terminal/session/leadTerminal parameter: the pane "
+                        + "to roll is always resolved from YOUR OWN connection, never a value you "
+                        + "pass, so you can only ever roll yourself — never another lead. Requires "
+                        + "leadRollover: to be configured; when it is not, every action returns a "
+                        + "clean refusal naming NOT_CONFIGURED instead of failing.",
                 objectSchema(Map.of(
-                        "action", stringProp("\"open\", \"confirm\" or \"cancel\""),
+                        "action", stringProp("\"open\", \"confirm\", \"cancel\" or \"status\""),
                         "reason", stringProp("Free-text audit note for \"open\" (optional, logged only)"),
-                        "token", stringProp("The token \"open\" returned — required for \"confirm\" and \"cancel\""),
+                        "token", stringProp("The token \"open\" returned — required for \"confirm\", "
+                                + "\"cancel\" and \"status\""),
                         "operatorConfirmed", Map.of("type", "boolean",
                                 "description", "For \"confirm\": your answer to \"has the human operator "
                                         + "confirmed this wipe\" (default false; only consulted when "

fleetd/src/test/java/dev/ltms/fleet/lead/LeadRolloverTest.java

@@ -16,6 +16,7 @@ import org.junit.jupiter.api.io.TempDir;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
@@ -1004,6 +1005,324 @@ class LeadRolloverTest {
         }
     }
 
+    // ---- CB-... : LeadRollover#status makes the outcome of a confirmed roll readable -----------
+
+    @Test
+    @DisplayName("[STATUS 1] after the calling turn never settles, status() reports "
+            + "TURN_NEVER_SETTLED for that token — this assertion could not even be written before "
+            + "status() existed")
+    void statusReportsTurnNeverSettledAfterTheRollIsAbandoned() throws IOException {
+        FakeHerdr herdr = new FakeHerdr();
+        herdr.agentStatus("working"); // the calling lead's own pane — never goes idle in this test
+        Path handover = writeHandover("handover contents");
+        FleetConfig.LeadRollover config =
+                new FleetConfig.LeadRollover(handover.toString(), true, 3600, 1 /*turnSettleSeconds*/, 20, "text");
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(herdr, config, () -> clock.addAndGet(500));
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "every synchronous gate should pass; the refusal happens "
+                + "only inside the deferred continuation, which this test's synchronous runner has "
+                + "already run to completion by the time confirm() returns");
+        assertEquals(0, promptCallCount(herdr), "sanity: /clear was never sent");
+
+        LeadRollover.RollStatus status = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.TURN_NEVER_SETTLED, status.state());
+        assertTrue(status.detail().contains("turnSettleSeconds"), "the detail must name the knob a "
+                + "reader needs to raise: " + status.detail());
+        assertTrue(status.detail().toLowerCase().contains("no /clear"), "the detail must say plainly "
+                + "that no /clear was ever sent: " + status.detail());
+    }
+
+    @Test
+    @DisplayName("[STATUS 2] a roll that completes reports ROLLED for its token")
+    void statusReportsRolledForACompletedRoll() throws IOException {
+        FakeHerdr herdr = new FakeHerdr(); // default idle — a full successful roll
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(herdr, cfg(handover.toString()), fixedClock(clock));
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "expected approval; got: " + decision.reason() + " / " + decision.detail());
+        assertEquals(2, promptCallCount(herdr), "sanity: /clear then bootstrapText were both sent");
+
+        LeadRollover.RollStatus status = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.ROLLED, status.state());
+        assertNotNull(status.detail());
+    }
+
+    @Test
+    @DisplayName("[STATUS 3] a roll where /clear never settles reports CLEAR_NEVER_SETTLED — "
+            + "distinct from both ROLLED and TURN_NEVER_SETTLED")
+    void statusReportsClearNeverSettledDistinctFromTheOtherTwoStates() throws IOException {
+        // Idle until /clear is sent, then permanently working — the SECOND wait never settles.
+        FakeHerdr fake = new FakeHerdr();
+        HerdrClient flipsAfterClear = new HerdrClient() {
+            @Override
+            public JsonNode call(String method, Object params) throws HerdrException {
+                JsonNode result = fake.call(method, params);
+                if ("agent.prompt".equals(method) && String.valueOf(params).contains("/clear")) {
+                    fake.agentStatus("working");
+                }
+                return result;
+            }
+
+            @Override
+            public void close() {
+                fake.close();
+            }
+        };
+        Path handover = writeHandover("handover contents");
+        FleetConfig.LeadRollover config =
+                new FleetConfig.LeadRollover(handover.toString(), true, 3600, 20, 1 /*clearSettleSeconds*/, "boot text");
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(flipsAfterClear, config, () -> clock.addAndGet(500));
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "every synchronous gate passes; the refusal is logged only, "
+                + "deep inside the deferred continuation");
+        assertEquals(1, promptCallCount(fake), "sanity: only /clear was sent, never bootstrapText");
+
+        LeadRollover.RollStatus status = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.CLEAR_NEVER_SETTLED, status.state());
+        assertNotEquals(LeadRollover.RollState.ROLLED, status.state());
+        assertNotEquals(LeadRollover.RollState.TURN_NEVER_SETTLED, status.state());
+        assertTrue(status.detail().contains("clearSettleSeconds"), status.detail());
+    }
+
+    @Test
+    @DisplayName("[STATUS 4] a token that was never issued, or was cancelled, gives a clean "
+            + "UNKNOWN answer rather than an exception or a false ROLLED")
+    void statusOnUnissuedOrCancelledTokenIsCleanNotAnExceptionOrFalseSuccess() throws IOException {
+        FakeHerdr herdr = new FakeHerdr();
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(herdr, cfg(handover.toString()), fixedClock(clock));
+
+        // never issued at all
+        LeadRollover.RollStatus neverIssued = assertDoesNotThrow(() -> rollover.status("no-such-token"));
+        assertEquals(LeadRollover.RollState.UNKNOWN, neverIssued.state());
+        assertNotEquals(LeadRollover.RollState.ROLLED, neverIssued.state());
+
+        // null / blank must not throw either — pending is a ConcurrentHashMap, which throws on a
+        // null-key lookup unless status() guards it first
+        assertEquals(LeadRollover.RollState.UNKNOWN, assertDoesNotThrow(() -> rollover.status(null)).state());
+        assertEquals(LeadRollover.RollState.UNKNOWN, assertDoesNotThrow(() -> rollover.status("  ")).state());
+
+        // opened, then cancelled
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        assertTrue(rollover.cancel(pending.token()));
+        LeadRollover.RollStatus cancelled = assertDoesNotThrow(() -> rollover.status(pending.token()));
+        assertEquals(LeadRollover.RollState.UNKNOWN, cancelled.state());
+        assertNotEquals(LeadRollover.RollState.ROLLED, cancelled.state());
+    }
+
+    @Test
+    @DisplayName("[STATUS 5] the bounded outcome history never grows past its cap")
+    void statusHistoryDoesNotGrowPastItsCap() throws IOException {
+        FakeHerdr herdr = new FakeHerdr(); // default idle — every roll completes
+        Path handover = writeHandover("handover contents"); // one file, reused by every roll below —
+        // checkHandover only compares its mtime against each open()'s OWN requestedAtMillis (the
+        // fake clock, in the low thousands), and the file's real (wall-clock) mtime is always far
+        // larger than that, so freshness passes on every iteration without rewriting the file.
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(herdr, cfg(handover.toString()), () -> clock.addAndGet(1));
+
+        int rolls = LeadRollover.OUTCOME_HISTORY_CAP + 50;
+        String[] tokens = new String[rolls];
+        for (int i = 0; i < rolls; i++) {
+            LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full #" + i);
+            LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+            assertTrue(decision.accepted(), "roll #" + i + " should have been approved: " + decision.detail());
+            tokens[i] = pending.token();
+        }
+
+        assertEquals(LeadRollover.RollState.ROLLED, rollover.status(tokens[rolls - 1]).state(),
+                "the most recently finished roll's outcome must still be in the bounded history");
+
+        // There is no direct size accessor for the bounded history, so boundedness is asserted
+        // indirectly and behaviourally: the OLDEST finished roll's outcome must have been evicted
+        // (reads back as a clean UNKNOWN, exactly like a token that was never issued) once more than
+        // OUTCOME_HISTORY_CAP rolls have gone through this instance. If the cap were not enforced,
+        // tokens[0] would still read back ROLLED here, and this assertion would fail.
+        assertEquals(LeadRollover.RollState.UNKNOWN, rollover.status(tokens[0]).state(),
+                "the oldest finished roll's outcome must have been evicted once the cap was "
+                        + "exceeded — otherwise the bounded history is not actually bounded");
+    }
+
+    @Test
+    @DisplayName("[STATUS 6] a status() call against a pending roll is read-only — no herdr call is "
+            + "made and the pending request is left untouched")
+    void statusCallAgainstAPendingRollIsReadOnlyAndTouchesNothing() throws IOException {
+        FakeHerdr herdr = new FakeHerdr();
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        LeadRollover rollover = newRollover(herdr, cfg(handover.toString()), fixedClock(clock));
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+
+        LeadRollover.RollStatus status = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.PENDING, status.state());
+        assertEquals(0, herdr.calls.size(), "status() must never make any herdr call at all — not "
+                + "just no agent.prompt — since it must never schedule, cancel, or retry anything");
+
+        // the pending request must be left exactly as it was: the SAME token can still be confirmed
+        // afterwards, as if status() had never been called.
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "status() must not have consumed or otherwise disturbed the "
+                + "pending request: " + decision.reason() + " / " + decision.detail());
+    }
+
+    // ---- PR #600 review round 2: IN_PROGRESS — the gap between confirm() handing off and the ----
+    // ---- continuation finishing must never read back as UNKNOWN ("nothing was ever requested") --
+
+    /**
+     * A {@code continuationRunner} that CAPTURES the roll instead of running it, so a test can
+     * observe {@link LeadRollover#status} in the window between {@link LeadRollover#confirm}
+     * handing off and the roll actually finishing — the window the synchronous {@code
+     * Runnable::run} runner used everywhere else in this class collapses to nothing. Call {@link
+     * #runNext()} to finish exactly one held roll, once the test is done observing the in-flight
+     * state.
+     */
+    private static final class HoldingRunner implements java.util.function.Consumer<Runnable> {
+        private final List<Runnable> held = new ArrayList<>();
+
+        @Override
+        public void accept(Runnable runnable) {
+            held.add(runnable);
+        }
+
+        int heldCount() {
+            return held.size();
+        }
+
+        /** Runs (and removes) the oldest held roll — FIFO, matching confirm() call order. */
+        void runNext() {
+            held.remove(0).run();
+        }
+    }
+
+    private static LeadRollover newRolloverWithHoldingRunner(HerdrClient herdr,
+            FleetConfig.LeadRollover config, LongSupplier nowMillis, HoldingRunner runner) {
+        AgentControl agents = new AgentControl(herdr);
+        return new LeadRollover(agents, () -> config, _ -> null, nowMillis, () -> { }, runner);
+    }
+
+    @Test
+    @DisplayName("[IN-PROGRESS 1] between an approved confirm() and the continuation finishing, "
+            + "status() reports IN_PROGRESS — not UNKNOWN, and not PENDING")
+    void statusReportsInProgressBetweenConfirmAndTheContinuationFinishing() throws IOException {
+        FakeHerdr herdr = new FakeHerdr(); // default idle — the held roll WOULD complete once run
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        HoldingRunner runner = new HoldingRunner();
+        LeadRollover rollover = newRolloverWithHoldingRunner(herdr, cfg(handover.toString()),
+                fixedClock(clock), runner);
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "expected approval; got: " + decision.reason() + " / " + decision.detail());
+        assertEquals(1, runner.heldCount(), "sanity: the roll must have been handed to the "
+                + "continuation runner and held there, not run yet");
+        assertEquals(0, promptCallCount(herdr), "sanity: the held continuation has not run, so "
+                + "nothing has been sent to the pane yet");
+
+        LeadRollover.RollStatus status = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.IN_PROGRESS, status.state(),
+                "a lead calling status() right after confirm() returned approved, while the roll "
+                        + "is still running, must be told IN_PROGRESS — not UNKNOWN (\"nothing was "
+                        + "ever requested\", which would wrongly invite it to call open() again "
+                        + "mid-roll) and not PENDING (\"not yet approved\", which is simply false "
+                        + "here): got " + status.state() + " / " + status.detail());
+    }
+
+    @Test
+    @DisplayName("[IN-PROGRESS 2] once the continuation finishes, the same token reports its "
+            + "terminal state — IN_PROGRESS is not sticky")
+    void inProgressStateIsNotStickyOnceTheContinuationFinishes() throws IOException {
+        FakeHerdr herdr = new FakeHerdr(); // default idle — the held roll completes once run
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        HoldingRunner runner = new HoldingRunner();
+        LeadRollover rollover = newRolloverWithHoldingRunner(herdr, cfg(handover.toString()),
+                fixedClock(clock), runner);
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "expected approval; got: " + decision.reason() + " / " + decision.detail());
+        assertEquals(LeadRollover.RollState.IN_PROGRESS, rollover.status(pending.token()).state(),
+                "sanity: must be IN_PROGRESS before the held continuation is run");
+
+        runner.runNext(); // finish the held roll now
+
+        assertEquals(2, promptCallCount(herdr), "sanity: the roll actually ran to completion "
+                + "once released — /clear then bootstrapText");
+        LeadRollover.RollStatus after = rollover.status(pending.token());
+        assertEquals(LeadRollover.RollState.ROLLED, after.state(),
+                "the SAME token must now report its terminal state — IN_PROGRESS must not still be "
+                        + "reported once the roll has actually finished");
+    }
+
+    @Test
+    @DisplayName("[IN-PROGRESS 4] IN_PROGRESS is distinct from every other RollState")
+    void inProgressStateIsDistinctFromAllOtherStates() throws IOException {
+        FakeHerdr herdr = new FakeHerdr();
+        Path handover = writeHandover("handover contents");
+        AtomicLong clock = new AtomicLong(1_000);
+        HoldingRunner runner = new HoldingRunner();
+        LeadRollover rollover = newRolloverWithHoldingRunner(herdr, cfg(handover.toString()),
+                fixedClock(clock), runner);
+
+        LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full");
+        LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+        assertTrue(decision.accepted(), "expected approval; got: " + decision.reason() + " / " + decision.detail());
+
+        LeadRollover.RollState inProgress = rollover.status(pending.token()).state();
+        assertEquals(LeadRollover.RollState.IN_PROGRESS, inProgress);
+        for (LeadRollover.RollState other : LeadRollover.RollState.values()) {
+            if (other == LeadRollover.RollState.IN_PROGRESS) {
+                continue;
+            }
+            assertNotEquals(other, inProgress, "IN_PROGRESS must be distinct from " + other);
+        }
+    }
+
+    @Test
+    @DisplayName("[IN-PROGRESS 5] eviction counts IN_PROGRESS entries toward the cap exactly like "
+            + "finished ones — a burst of confirmed-but-not-yet-finished rolls still ages out")
+    void evictionCountsInProgressEntriesTowardTheCap() throws IOException {
+        FakeHerdr herdr = new FakeHerdr();
+        Path handover = writeHandover("handover contents"); // reused by every roll — see
+        // statusHistoryDoesNotGrowPastItsCap for why one shared file is enough for freshness.
+        AtomicLong clock = new AtomicLong(1_000);
+        HoldingRunner runner = new HoldingRunner(); // nothing run below — every roll stays IN_PROGRESS
+        LeadRollover rollover = newRolloverWithHoldingRunner(herdr, cfg(handover.toString()),
+                () -> clock.addAndGet(1), runner);
+
+        int rolls = LeadRollover.OUTCOME_HISTORY_CAP + 50;
+        String[] tokens = new String[rolls];
+        for (int i = 0; i < rolls; i++) {
+            LeadRollover.PendingRollover pending = rollover.open(LEAD, "context is full #" + i);
+            LeadRollover.RollDecision decision = rollover.confirm(LEAD, pending.token(), true);
+            assertTrue(decision.accepted(), "roll #" + i + " should have been approved: " + decision.detail());
+            tokens[i] = pending.token();
+        }
+        assertEquals(rolls, runner.heldCount(), "sanity: none of these rolls have been run — every "
+                + "one of them is sitting in outcomes as IN_PROGRESS, not a separate uncapped map");
+
+        assertEquals(LeadRollover.RollState.IN_PROGRESS, rollover.status(tokens[rolls - 1]).state(),
+                "the most recently confirmed (still in-flight) roll must still be in the bounded "
+                        + "history");
+        assertEquals(LeadRollover.RollState.UNKNOWN, rollover.status(tokens[0]).state(),
+                "the oldest confirmed roll's IN_PROGRESS entry must have been evicted once the cap "
+                        + "was exceeded, exactly like a finished entry would be — proving IN_PROGRESS "
+                        + "entries share the SAME bounded map and count against the SAME cap, rather "
+                        + "than living in a second, uncapped in-flight map");
+    }
+
     @Test
     @DisplayName("[fleetd #494 follow-up] the turn-settle timeout warn line prints the MEASURED "
             + "elapsed time next to the configured budget, never the configured value alone")

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

@@ -258,5 +258,51 @@ class FleetMcpHandoverTest {
         assertTrue(textOf(r).contains("\"cancelled\":true"), textOf(r));
     }
 
+    // --- new: action "status" — makes the outcome of a confirm() readable through the tool ----
+
+    @Test
+    @DisplayName("status on a null LeadRollover is a clean NOT_CONFIGURED refusal, never a throw")
+    void statusWithNullLeadRolloverRefusesCleanly() {
+        McpSchema.CallToolResult r = assertDoesNotThrow(() -> FleetMcp.handover(null, LEAD,
+                Map.of("action", "status", "token", "whatever")));
+        assertFalse(r.isError());
+        assertTrue(textOf(r).contains("NOT_CONFIGURED"), textOf(r));
+    }
+
+    @Test
+    @DisplayName("status on a token that was never opened reports UNKNOWN")
+    void statusOnUnknownTokenReportsUnknown() {
+        LeadRollover rollover = newRollover(tmp.resolve("h.md").toString());
+
+        McpSchema.CallToolResult r = FleetMcp.handover(rollover, LEAD,
+                Map.of("action", "status", "token", "does-not-exist"));
+        assertFalse(r.isError());
+        assertTrue(textOf(r).contains("\"state\":\"UNKNOWN\""), textOf(r));
+    }
+
+    @Test
+    @DisplayName("status on a token that is still pending (opened, not confirmed) reports PENDING")
+    void statusOnPendingTokenReportsPending() {
+        LeadRollover rollover = newRollover(tmp.resolve("h.md").toString());
+        String token = extractToken(textOf(FleetMcp.handover(rollover, LEAD, Map.of("action", "open"))));
+
+        McpSchema.CallToolResult r = FleetMcp.handover(rollover, LEAD,
+                Map.of("action", "status", "token", token));
+        assertFalse(r.isError());
+        assertTrue(textOf(r).contains("\"state\":\"PENDING\""), textOf(r));
+    }
+
+    @Test
+    @DisplayName("status is registered on the tool's schema and the schema still names no caller-identity parameter")
+    void statusActionIsAdvertisedOnTheSchema() {
+        FleetMcp m = mcp(null);
+        McpSchema.Tool tool = m.registeredTools().stream()
+                .filter(t -> "fleet_handover".equals(t.name()))
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("fleet_handover was not registered"));
+        assertTrue(tool.description().contains("'status'"),
+                "the tool's own description must advertise the 'status' action: " + tool.description());
+    }
+
     // --- acceptance 7 (wiring) is covered by FleetdLeadRolloverWiringTest, unchanged -----------
 }

scripts/redeploy-fleetd.sh

@@ -50,6 +50,14 @@
 #      absence is the real signal, because a drain that dies on its first session prints nothing
 #      else either. Warns loudly; never fails the redeploy, because by the time this is detectable
 #      the new daemon is already up and healthy.
+#  10. fleetd #603 — the same shape as trap 3 above, through a different door: the step that waited
+#      for the NEW process to appear gave it its own short, fixed 10s budget, then hard-`die`d,
+#      while the health check right after it waits a full $HEALTH_WAIT (60s) for the same daemon to
+#      answer. Under launchd, `launchctl load` returns as soon as launchd accepts the job, before the
+#      java process exists, and on a slow host that took longer than 10s — so the script died with
+#      "no process appeared" on a deploy that had fully succeeded. The pid poll now shares
+#      $HEALTH_WAIT instead of a separate, shorter budget, and a miss there falls through to the
+#      health check (the truer signal: is it actually answering?) instead of killing the run.
 #
 # Usage:
 #   scripts/redeploy-fleetd.sh              # build, confirm, restart, verify
@@ -79,7 +87,9 @@ OUT="$MODULE/fleetd.out"
 PATTERN='target/fleetd.jar'
 HEALTH='http://127.0.0.1:8765/healthz'
 STOP_WAIT=30      # seconds to wait for a clean exit before reporting failure
-HEALTH_WAIT=60    # seconds to wait for /healthz to answer after start
+HEALTH_WAIT=60    # seconds to wait for /healthz to answer after start — fleetd #603: also the pid-
+                  # poll budget below (wait_for_new_pid/await_daemon_started), so the two checks
+                  # share one named budget instead of the pid poll holding its own shorter one
 
 # CB-594: the launchd agent this script must not fight with (see trap 6 above).
 LAUNCHD_LABEL='dev.ltms.fleetd'
@@ -1067,6 +1077,67 @@ $(tail -30 "$out_file" 2>/dev/null)"
   fi
 }
 
+# fleetd #603 — the dual of wait_for_daemon_exit above: waits for a pid to APPEAR instead of
+# disappear. Used to be a bare `for _ in $(seq 10)` sitting directly in the main flow, with its own
+# short, fixed budget that had nothing to do with $HEALTH_WAIT (60s) — the budget the health check
+# right after it gets for the very same daemon. Under launchd, `launchctl load` returns as soon as
+# launchd accepts the job, before the java process exists, and on a slow host that took longer than
+# 10s — so the script died with "no process appeared" on a deploy that had fully succeeded (the
+# operator confirmed pid, healthz, and a fresh log line all present, by hand, right afterwards).
+# Sharing $HEALTH_WAIT here removes the extra, shorter magic number without inventing a new one.
+wait_for_new_pid() {
+  local timeout="$1" _i
+  for _i in $(seq "$timeout"); do
+    [ -n "$(running_pid)" ] && return 0
+    sleep 1
+  done
+  [ -n "$(running_pid)" ]
+}
+
+# fleetd #603 — the pid-appeared check and the healthz check, folded into one decision. Same shape
+# as swap_if_built/refuse_drain_gate/report_shutdown_drain above (#521/#528/#512): the main flow
+# calls this ONE function unconditionally, so there is no bare guard left for a future edit to
+# invert independently of it. That matters more here than for most of those: a plain grep of this
+# script's source cannot tell "a pid miss falls through to the health check" from "a pid miss still
+# dies" apart, because both read as the same two lines of text with only the runtime branch
+# changed — a source-text test could pass on either behavior. A real behavioural test on this
+# function is the only thing that can actually tell them apart, which is why one exists below.
+#
+# wait_for_new_pid's miss is no longer fatal by itself: it falls through to the health check, which
+# is direct proof the new daemon is up (/healthz answers 200) rather than a proxy for it (a process
+# merely existing under a name running_pid() recognises). A genuine failure still dies here: it
+# misses the pid poll AND the health poll, and report_health's own die() still prints the tail of
+# $out_file, exactly as before this fix.
+#
+# Sets NEW_PID (global — the caller's "pid N, jar ..." result line reads it afterwards) and
+# HEALTH_BODY/HEALTH_CODE (globals, the same reason report_health already needs them handed back).
+# Only ever called as a bare statement in the main flow below, never from inside a `$( )`: a die()
+# reached from inside a command substitution only kills that subshell, not the whole script, which
+# would silently turn a genuine failure back into a false "succeeded" exit (see poll_health_body's
+# own `|| true` idiom for the same hazard from the other direction).
+await_daemon_started() {
+  local health_wait="$1" old_pid="$2" health_url="$3" out_file="$4"
+  NEW_PID=""
+  if wait_for_new_pid "$health_wait"; then
+    NEW_PID="$(running_pid)"
+    [ "$NEW_PID" != "${old_pid:-}" ] || die "pid unchanged ($NEW_PID) — the old daemon never died"
+    ok "started, pid $NEW_PID"
+  else
+    warn "no process matched $PATTERN within ${health_wait}s of starting — falling through to the health check, which is the more truthful signal"
+  fi
+
+  HEALTH_BODY="$(poll_health_body "$health_url" "$health_wait")" || true
+  HEALTH_CODE="000"
+  [ -n "$HEALTH_BODY" ] || HEALTH_CODE="$(curl -s -o /dev/null -w '%{http_code}' --max-time 2 "$health_url" 2>/dev/null || echo 000)"
+  report_health "$HEALTH_BODY" "$HEALTH_CODE" "$out_file" "$health_wait"
+
+  # By now /healthz has answered (report_health above would have died otherwise), so the daemon is
+  # confirmed up even if the pid poll never matched it — see running_pid()'s own comment on
+  # under-counting if the launch method ever stops being a plain `java -jar`. Fill NEW_PID in for the
+  # result line rather than leave it blank on an otherwise fully successful redeploy.
+  [ -n "$NEW_PID" ] || NEW_PID="$(running_pid)"
+}
+
 # Ticket item 8 — `HAD_OLD_PID=0; [ -n "$OLD_PID" ] && HAD_OLD_PID=1`. Measured safe under `set -e`
 # at both bash 3.2.57 and 5.x (see the header comment trap 9 discussion in the ticket) — not a `set
 # -e` hazard, but still an untested computation feeding report_shutdown_drain's own four-way
@@ -1285,28 +1356,14 @@ say "start"
 # there now, so this line is the only thing left in the main flow to get wrong.
 dispatch_start "$SUPERVISOR_KIND"
 
-for _ in $(seq 10); do
-  NEW_PID="$(running_pid)"
-  [ -n "$NEW_PID" ] && break
-  sleep 1
-done
-[ -n "${NEW_PID:-}" ] || die "no process appeared. Last lines of $OUT:
-$(tail -20 "$OUT" 2>/dev/null)"
-[ "$NEW_PID" != "${OLD_PID:-}" ] || die "pid unchanged ($NEW_PID) — the old daemon never died"
-ok "started, pid $NEW_PID"
-
 # ------------------------------------------------------------------ verify
 
 say "verify"
 
-# fleetd #555: poll_health_body/health_is_up/report_health above. HEALTH_CODE is only ever
-# consulted by report_health when the body came back empty; `|| true` on both assignments is the
-# same "an absent/failing command substitution must not kill the script under set -e" idiom the
-# swap/drain helpers already rely on (see poll_health_body's own comment).
-HEALTH_BODY="$(poll_health_body "$HEALTH" "$HEALTH_WAIT")" || true
-HEALTH_CODE="000"
-[ -n "$HEALTH_BODY" ] || HEALTH_CODE="$(curl -s -o /dev/null -w '%{http_code}' --max-time 2 "$HEALTH" 2>/dev/null || echo 000)"
-report_health "$HEALTH_BODY" "$HEALTH_CODE" "$OUT" "$HEALTH_WAIT"
+# fleetd #603: await_daemon_started above folds the pid-appeared check and the healthz check into
+# one decision — see its own comment for why a bare `if` here, split back into the old two pieces,
+# would put back an untestable branch this ticket exists to close.
+await_daemon_started "$HEALTH_WAIT" "$OLD_PID" "$HEALTH" "$OUT"
 
 # A fresh listening line, strictly after the restart mark. An old daemon that never died would
 # otherwise let an old line pass for a new one.
@@ -1361,7 +1418,7 @@ report_shutdown_drain "$FRESH_LOG" "$HAD_OLD_PID"
 assert_single_daemon "$(running_pid)"
 
 say "result"
-ok "pid $NEW_PID, jar $(jar_id)"
+ok "pid ${NEW_PID:-unknown}, jar $(jar_id)"
 if [ "$REDEPLOY_AMQP_CHECK_SKIPPED" -eq 1 ]; then
   # fleetd #552: the fourth reader of the fresh-log region. Without this branch REDEPLOY_ERROR_COUNT
   # stays at its untouched 0 (classify_amqp_connection_errors never got a file to read) and

scripts/test-redeploy-fleetd.sh

@@ -770,6 +770,137 @@ test_wait_for_daemon_exit_times_out_if_pid_never_clears() {
   source "$ROOT/scripts/redeploy-fleetd.sh"   # restore the real running_pid/sleep for later tests
 }
 
+# fleetd #603 — wait_for_new_pid is the dual of wait_for_daemon_exit above: it must not report
+# success while running_pid() still answers empty, and must report success the moment a pid
+# appears. Same counter-file idiom as test_wait_for_daemon_exit_returns_true_once_pid_clears above,
+# for the same reason (running_pid() runs inside a `$(...)` subshell on every call).
+test_wait_for_new_pid_returns_true_once_pid_appears() {
+  local counter_file="$TMP/wait-new-pid-calls" final_calls
+  printf '0' > "$counter_file"
+  running_pid() {
+    local n
+    n="$(cat "$counter_file")"
+    n=$((n + 1))
+    printf '%s' "$n" > "$counter_file"
+    if [ "$n" -lt 3 ]; then printf ''; else printf '4242'; fi
+  }
+  sleep() { :; }
+  wait_for_new_pid 10 || fail "wait_for_new_pid did not report success once the pid appeared"
+  final_calls="$(cat "$counter_file")"
+  [ "$final_calls" -ge 3 ] || fail "wait_for_new_pid returned before actually re-checking running_pid"
+  source "$ROOT/scripts/redeploy-fleetd.sh"   # restore the real running_pid/sleep for later tests
+}
+
+test_wait_for_new_pid_times_out_if_pid_never_appears() {
+  local rc=0
+  running_pid() { printf ''; }
+  sleep() { :; }
+  wait_for_new_pid 3 || rc=$?
+  [ "$rc" -ne 0 ] || fail "wait_for_new_pid reported success while the pid never appeared"
+  source "$ROOT/scripts/redeploy-fleetd.sh"   # restore the real running_pid/sleep for later tests
+}
+
+# fleetd #603 — await_daemon_started folds the pid-appeared check and the healthz check into one
+# decision (see its own comment in redeploy-fleetd.sh for why a source-text grep cannot tell the two
+# possible behaviors apart here). These two tests are the ticket's own acceptance criteria, run
+# together in this one suite invocation so neither can be satisfied by code that never fails at all:
+#
+#   1. a slow start must still succeed — running_pid mimics a process that does not appear until
+#      well after the OLD, buggy 10-second budget, but does appear, and healthz answers.
+#   2. a genuine failure must still fail, and still print the log tail — running_pid and the health
+#      check both report nothing at all, ever.
+test_await_daemon_started_slow_pid_then_healthy_succeeds() {
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+  stub_die_recorder
+  local counter_file="$TMP/await-slow-pid-calls" output
+  printf '0' > "$counter_file"
+  running_pid() {
+    local n
+    n="$(cat "$counter_file")"
+    n=$((n + 1))
+    printf '%s' "$n" > "$counter_file"
+    # Stays empty well past the old 10-second budget, then appears — the exact shape #603 reports.
+    if [ "$n" -lt 12 ]; then printf ''; else printf '4242'; fi
+  }
+  sleep() { :; }
+  poll_health_body() { printf '{"status":"ok"}'; return 0; }
+  # NOT `output="$(await_daemon_started ...)"`: that would run the call in a subshell, and
+  # NEW_PID — a plain global assignment inside the function, by design (see its own comment) — would
+  # die with that subshell instead of reaching this test's own shell. Redirect to a file instead, the
+  # same hazard await_daemon_started's own comment warns die() itself is subject to.
+  await_daemon_started 60 "" "http://ignored/healthz" "$TMP/await-slow-pid.out" \
+    > "$TMP/await-slow-pid-output.log" 2>&1
+  output="$(cat "$TMP/await-slow-pid-output.log")"
+  [ "$DIED_CALLED" = 0 ] \
+    || fail "await_daemon_started must not die on a slow-but-real start: $DIED_MESSAGE"
+  assert_equals "4242" "$NEW_PID" "await_daemon_started NEW_PID after a slow-but-real start"
+  printf '%s' "$output" | grep -qF 'started, pid 4242' \
+    || fail "await_daemon_started did not report the pid once it finally appeared"
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+}
+
+test_await_daemon_started_never_appears_dies_with_log_tail() {
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+  stub_die_recorder
+  local out_file="$TMP/await-never-appears.out"
+  printf 'boot line one\nboot line two\n' > "$out_file"
+  running_pid() { printf ''; }
+  sleep() { :; }
+  poll_health_body() { return 1; }
+  await_daemon_started 2 "" "http://127.0.0.1:1/healthz" "$out_file" > /dev/null 2>&1
+  [ "$DIED_CALLED" = 1 ] \
+    || fail "await_daemon_started must die when the daemon never appears and never becomes healthy"
+  printf '%s' "$DIED_MESSAGE" | grep -qF 'never answered' \
+    || fail "await_daemon_started die message does not say healthz never answered"
+  printf '%s' "$DIED_MESSAGE" | grep -qF 'boot line two' \
+    || fail "await_daemon_started die message does not include the log tail"
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+}
+
+# fleetd #603 review — the path the fall-through actually exists for, and the one gap a lead
+# mutation found in the first version of this test file: running_pid() NEVER finds anything (as its
+# own doc comment says it eventually will, once the daemon stops being launched as a plain
+# `java -jar` its allowlist recognises), while /healthz answers anyway. Neither of the two tests
+# above drives this: the slow-pid test has the pid appear, so the `else` branch never runs, and the
+# never-appears test fails BOTH checks, so it dies either way and cannot tell which branch fired.
+# This must not die, must warn (so the operator is told the pid could not be identified), and must
+# leave NEW_PID empty — the honest "could not establish this" answer, never a guessed pid, which is
+# what the final result line's `${NEW_PID:-unknown}` fallback exists to print truthfully.
+#
+# Proof this actually pins the behavior, not just the source text (paste from a real run, not
+# claimed): reverting the `warn` below back to `die "no process appeared"` (the old fleetd #603
+# defect, reintroduced) turns this one test red —
+#   FAIL: await_daemon_started must not die when the pid is never found but healthz answers
+# — and restoring `warn` turns the whole suite green again. Both halves observed, not asserted.
+test_await_daemon_started_pid_never_found_but_healthy_warns_and_survives() {
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+  stub_die_recorder
+  local out_file="$TMP/await-pid-never-found.out" output
+  printf 'boot line\n' > "$out_file"
+  running_pid() { printf ''; }
+  sleep() { :; }
+  poll_health_body() { printf '{"status":"ok"}'; return 0; }
+  # NOT `output="$(await_daemon_started ...)"` — see the slow-pid test above for why that would
+  # drop NEW_PID's assignment in a subshell instead of reaching this test's own shell.
+  await_daemon_started 3 "" "http://ignored/healthz" "$out_file" \
+    > "$TMP/await-pid-never-found-output.log" 2>&1
+  output="$(cat "$TMP/await-pid-never-found-output.log")"
+  [ "$DIED_CALLED" = 0 ] \
+    || fail "await_daemon_started must not die when the pid is never found but healthz answers: $DIED_MESSAGE"
+  printf '%s' "$output" | grep -qF 'falling through to the health check' \
+    || fail "await_daemon_started did not warn that the pid could not be identified"
+  assert_equals "" "$NEW_PID" \
+    "await_daemon_started NEW_PID when the pid is never found but healthz answers — must stay empty, never a guessed pid"
+  source "$ROOT/scripts/redeploy-fleetd.sh"
+}
+
+test_await_daemon_started_call_site_present() {
+  local src="$ROOT/scripts/redeploy-fleetd.sh" call_line
+  call_line="$(grep -Fn 'await_daemon_started "$HEALTH_WAIT" "$OLD_PID" "$HEALTH" "$OUT"' "$src" | head -1 | cut -d: -f1 || true)"
+  [ -n "$call_line" ] \
+    || fail "could not find the main flow's await_daemon_started call site in redeploy-fleetd.sh"
+}
+
 # fleetd #521 — the swap step's guard, at two levels.
 #
 # The first two tests call the predicate should_swap() directly. They pin its logic, and that is all
@@ -1350,9 +1481,11 @@ test_poll_health_body_returns_nonzero_when_unreachable() {
 
 test_report_health_call_site_present() {
   local src="$ROOT/scripts/redeploy-fleetd.sh" call_line
-  call_line="$(grep -Fn 'report_health "$HEALTH_BODY" "$HEALTH_CODE" "$OUT" "$HEALTH_WAIT"' "$src" | head -1 | cut -d: -f1 || true)"
+  # fleetd #603 — moved from a literal main-flow call into await_daemon_started (see its own
+  # comment above for why); this now finds the call inside that function instead.
+  call_line="$(grep -Fn 'report_health "$HEALTH_BODY" "$HEALTH_CODE" "$out_file" "$health_wait"' "$src" | head -1 | cut -d: -f1 || true)"
   [ -n "$call_line" ] \
-    || fail "could not find the main flow's report_health call site in redeploy-fleetd.sh"
+    || fail "could not find await_daemon_started's report_health call site in redeploy-fleetd.sh"
 }
 
 # fleetd #555 item 8 — `HAD_OLD_PID=0; [ -n "$OLD_PID" ] && HAD_OLD_PID=1`. Measured safe under
@@ -2041,6 +2174,12 @@ test_require_no_build_jar_dies_when_absent
 test_require_no_build_jar_accepts_present_jar
 test_wait_for_daemon_exit_returns_true_once_pid_clears
 test_wait_for_daemon_exit_times_out_if_pid_never_clears
+test_wait_for_new_pid_returns_true_once_pid_appears
+test_wait_for_new_pid_times_out_if_pid_never_appears
+test_await_daemon_started_slow_pid_then_healthy_succeeds
+test_await_daemon_started_never_appears_dies_with_log_tail
+test_await_daemon_started_pid_never_found_but_healthy_warns_and_survives
+test_await_daemon_started_call_site_present
 test_swap_ordered_after_wait_and_before_start
 test_drain_gate_abort_message_says_no_no_build
 test_drain_gate_refusal_build_ran_staged_present