fleetd #608: make anAlreadyCollectedTicketProducesNoNudge deterministic

Replace the real ScheduledExecutorService backing ReplyPushLoop in this one
test with ManualScheduler, a fake that only runs a tick when the test calls
runDueTasks(). The old test bet a 300ms backoff was wide enough that
collecting the ticket always won the race against the scheduler's own timer
- true on an idle machine, false under a loaded full-suite run, which is
exactly the flake reported.

The rewritten test also waits on setAfterFinishAsyncTaskCompleteHookForTest
(already used elsewhere in this file) instead of polling Phase.DONE, so it
does not race CompletableFuture.complete()'s own publish-then-run-dependents
gap (fleetd #399) while proving ReplyPushLoop.onTicketTerminal really ran
before the ticket is collected.

Verified: backoff=1 (the most hostile value) still passes; mutating
ReplyPushLoop.ticketCollected to a no-op turns the test red with the same
assertion message the original flake reported; three consecutive full-suite
runs are green (1841/1841 each).

CLAUDE.md

@@ -145,9 +145,10 @@ authorized to settle it, not only to advise. Architects first form independent p
 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.
+**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.
 
 | Intent | Tool |
 |---|---|

fleetd/src/main/java/dev/ltms/fleet/Fleetd.java

@@ -672,6 +672,10 @@ public final class Fleetd {
                 leadMailbox,
                 outageSource,
                 new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)),
+                // fleetd #602 gauge-wiring: threads each lead's configured configDir into the
+                // context gauge — see leadConfigDirSource's own doc for why this, not a hardcoded
+                // null, is what fleet_list's context row now reads.
+                leadConfigDirSource(() -> config.get().profiles(), leaders),
                 // fleetd #361: the operator-declared peers this daemon's fleet_list should try to
                 // reach. Read from the SAME snapshot leadMailbox itself opened from (cfg.coordinator()),
                 // not the live config.get() — coordinator wiring is already boot-time-fixed (see
@@ -1576,6 +1580,58 @@ public final class Fleetd {
         };
     }
 
+    /**
+     * fleetd #602 gauge-wiring: per-lead-name factory for {@link FleetMcp.LeadConfigDirSource} — the
+     * {@code CLAUDE_CONFIG_DIR} the named lead's own profile runs on, so {@code LeadContextGauge}
+     * reads the transcript directory that lead's Claude Code process actually writes to, not always
+     * the built-in {@code <user.home>/.claude} default.
+     *
+     * <p>Follows the same {@code fleet.leaders.<name>.profile} link {@link #leadSeatLookup} already
+     * uses to find a lead's profile, one step further to that profile's own {@code configDir:}. A
+     * lead entry that names no {@code profile:}, or whose named profile is not configured, or whose
+     * profile sets no {@code configDir:} override, returns {@code null} — {@code LeadContextGauge}
+     * then falls back to its own default, exactly as before this ticket.
+     *
+     * @param profiles the live profile map, normally {@code () -> config.get().profiles()} in
+     *                 {@code main} — read live, like every other {@code configDir} lookup, so a
+     *                 config reload takes effect on the very next {@code fleet_list} call
+     * @param leaders  {@code fleet.leaders}, read once at startup like {@link #leadSeatLookup}'s own
+     *                 {@code leaders} parameter — passed as a plain map, never re-read from
+     *                 {@code config.get()}
+     */
+    static Function<String, String> leadConfigDirLookup(Supplier<Map<String, FleetConfig.Profile>> profiles,
+            Map<String, FleetConfig.Leader> leaders) {
+        return leadName -> {
+            FleetConfig.Leader lead = leaders.get(leadName);
+            if (lead == null || lead.profile() == null || lead.profile().isBlank()) {
+                return null;
+            }
+            FleetConfig.Profile leadProfile = profiles.get().get(lead.profile());
+            return leadProfile == null ? null : leadProfile.configDir();
+        };
+    }
+
+    /**
+     * fleetd #602 gauge-wiring follow-up (PR #606 review comment 17353): {@code main} used to build
+     * {@code new FleetMcp.LeadConfigDirSource(leadConfigDirLookup(...))} inline, with nothing a test
+     * could call directly. Measured on that shape: replacing the whole expression with {@code
+     * FleetMcp.LeadConfigDirSource.none()} at the call site compiled with 0 errors and left the full
+     * 1822-test suite green — the daemon could be changed to always report every lead's context as
+     * {@code UNKNOWN}, forever, while every test stayed green. That is the same hand-built-vs-wired
+     * shape as fleetd #561/#248/#426/#562 ({@link #loopHealthSource}).
+     *
+     * <p>The fix extracts the inline {@code new} into this factory, in the same style as {@link
+     * #loopHealthSource}/{@link #capacitySource}/{@link #healthCoverageSource} — which is exactly
+     * what makes it directly callable from {@code FleetdLeadConfigDirSourceWiringTest}. That test
+     * calls this factory with real {@link FleetConfig.Profile}/{@link FleetConfig.Leader} fixtures and
+     * asserts the returned source resolves a real {@code configDir} — a property that would be false
+     * if this method's body were mutated to {@code return FleetMcp.LeadConfigDirSource.none();}.
+     */
+    static FleetMcp.LeadConfigDirSource leadConfigDirSource(Supplier<Map<String, FleetConfig.Profile>> profiles,
+            Map<String, FleetConfig.Leader> leaders) {
+        return new FleetMcp.LeadConfigDirSource(leadConfigDirLookup(profiles, leaders));
+    }
+
     /**
      * fleetd #248 / fleetd#201 Unit 5: package-private factory for the per-target backend-error
      * pattern lookup {@link CompletionResolver} classifies a pane scrape against. Closes over the

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

@@ -0,0 +1,307 @@
+package dev.ltms.fleet.lead;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.LongSupplier;
+
+/**
+ * Reads how full a lead's own Claude Code context window is, from the transcript Claude Code
+ * itself writes — never from the lead's pane (fleetd has {@code AgentControl.read} for that, and
+ * this must not use it: a pane holds terminal text, not the structured usage numbers a transcript
+ * carries, and scraping it would also race the lead's own rendering).
+ *
+ * <p><strong>Why this exists.</strong> A lead auto-compacts when its context fills — on the host
+ * this was built for, that happened 30 times in one session, discarding roughly 250,000 tokens
+ * and costing 46 seconds to 3 minutes each time, and fleetd had no way to see it coming. This
+ * class is the first thing that looks.
+ *
+ * <p><strong>The route.</strong> Claude Code appends one JSON object per line to
+ * {@code <configDir>/projects/<slug>/<sessionId>.jsonl}. {@code <slug>} is an undocumented,
+ * internal encoding of the working directory — this class never derives it. Instead it lists the
+ * one-level-deep subdirectories of {@code <configDir>/projects/} and looks for
+ * {@code <sessionId>.jsonl} by name, so the slug rule can change without breaking this reader.
+ *
+ * <ul>
+ *   <li><strong>Live context</strong> is read off the last record in the read window that carries
+ *       a {@code message.usage} object: {@code input_tokens + cache_read_input_tokens +
+ *       cache_creation_input_tokens}. This is what actually fills the window — a plain
+ *       {@code input_tokens} count alone understates it once the conversation has any cached
+ *       prefix, which on a long-lived lead is always.</li>
+ *   <li><strong>Compaction history</strong> is a count of {@code subtype: "compact_boundary"}
+ *       records seen in the same read window — see {@link Reading#compactions()}. It is a count
+ *       within the window this reader actually looked at, not a lifetime total: a session with
+ *       more compactions than fit in {@link #TAIL_BYTES} of transcript will undercount. That
+ *       trade-off is deliberate — see {@link #TAIL_BYTES}.</li>
+ * </ul>
+ *
+ * <p><strong>Three states, not two (OK / HIGH / UNKNOWN).</strong> Every path that cannot
+ * positively establish the live token count — a missing file, an unreadable one, a peer that
+ * is not a Claude backend, or every line in the read window failing to parse as JSON — returns
+ * {@link State#UNKNOWN} with no token number, never a default "0" or "ok" that would read as
+ * "this lead is fine" when the honest answer is "I could not look".
+ *
+ * <p><strong>A torn final line does not mean UNKNOWN.</strong> {@code fleet_list} reads this
+ * transcript while Claude Code may be mid-write on it, so the last line in the window can be cut
+ * off mid-flush — that is an ordinary, expected race, not a sign the format has changed. Earlier
+ * this class treated ANY unparseable last line as UNKNOWN, on the theory that "if the format
+ * changes, we should see UNKNOWN". That reasoning does not hold: a real format change makes
+ * <em>every</em> line in the window unparseable, not only the last one written. So a single
+ * malformed line (most often the final, torn one, but the check is not position-specific) is
+ * simply skipped rather than treated as fatal, and the reading is built from whatever lines in the
+ * window did parse. Only when <em>none</em> of them parse — the real format-change signal — does
+ * this return {@link State#UNKNOWN}, still with no stale number standing in for "I could not
+ * tell".
+ *
+ * <p><strong>Bounded cost.</strong> {@code fleet_list} is polled constantly, so every read is
+ * capped two ways: {@link #TAIL_BYTES} bounds how much of the transcript is ever read from disk
+ * (never the whole 52 MB a long-lived transcript reaches on the host this was measured on), and
+ * {@link #DEFAULT_CACHE_TTL_MILLIS} bounds how often that bounded read actually happens — a burst
+ * of {@code fleet_list} calls inside one TTL window reads the file once. One instance's cache is
+ * keyed by {@code (configDir, sessionId)}, so it is safe to share across every lead a single
+ * {@code fleet_list} call reports on.
+ */
+public final class LeadContextGauge {
+
+    private static final String PROJECTS_DIR = "projects";
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    /**
+     * How many trailing bytes of a transcript a single read ever pulls off disk. Chosen so one
+     * read comfortably spans many recent turns — each usage or compact_boundary record is at most
+     * a few KB — while staying nowhere near the 52 MB a long session's real transcript reaches on
+     * the host this was built for; reading that whole file on every {@code fleet_list} call is
+     * exactly the cost this bound exists to avoid. 2 MiB holds on the order of hundreds of recent
+     * lines even when a turn's tool output is unusually large, which is far more than needed to
+     * find the most recent usage record and any recent compaction.
+     */
+    static final int TAIL_BYTES = 2 * 1024 * 1024;
+
+    /**
+     * How long a {@link Reading} is served from cache before the file is read again.
+     * {@code fleet_list} is called constantly (by design — it is the fleet's own status probe), so
+     * without a TTL a burst of calls would re-read the transcript tail once per call. 5 seconds is
+     * short enough that a caller watching for a state change never waits long, and long enough that
+     * a poll loop calling every second or two only touches disk once per window.
+     */
+    static final long DEFAULT_CACHE_TTL_MILLIS = 5_000;
+
+    /**
+     * Live tokens at or above this count report {@link State#HIGH}. On the host this was measured
+     * on, auto-compaction actually fires around 267,000–270,000 tokens, but the point of a HIGH
+     * state is to warn before that happens, not at it — 200,000 is the standard Claude context
+     * window size and a sensible built-in default: no config key is required to pick it, and a
+     * lead crossing it is already deep enough into its window that a compaction is foreseeable.
+     */
+    static final long HIGH_THRESHOLD_TOKENS = 200_000;
+
+    /** The only peer kind this reader understands ({@code Agent.agentType()}'s wire value). */
+    private static final String CLAUDE_AGENT_TYPE = "claude";
+
+    public enum State { OK, HIGH, UNKNOWN }
+
+    /**
+     * @param state       {@link State#UNKNOWN} whenever {@code tokens} could not be established
+     * @param tokens      live context tokens, or {@code null} exactly when {@code state} is
+     *                    {@link State#UNKNOWN}
+     * @param compactions {@code compact_boundary} records seen in the read window (see class
+     *                    javadoc) — {@code 0} both for "genuinely none seen" and for "unknown",
+     *                    since a caller that already sees {@code state: UNKNOWN} has no reason to
+     *                    trust this number either way
+     */
+    public record Reading(State state, Long tokens, int compactions) {
+        static Reading unknown() {
+            return new Reading(State.UNKNOWN, null, 0);
+        }
+    }
+
+    private record CacheEntry(Reading reading, long readAtMillis) {
+    }
+
+    private final LongSupplier clock;
+    private final long ttlMillis;
+    private final Map<String, CacheEntry> cache = new ConcurrentHashMap<>();
+    /** Test seam only (package-private) — counts real disk reads, i.e. cache misses. */
+    private final AtomicInteger diskReads = new AtomicInteger();
+
+    public LeadContextGauge() {
+        this(System::currentTimeMillis, DEFAULT_CACHE_TTL_MILLIS);
+    }
+
+    /** Test seam: an injectable clock and TTL so cache expiry is provable without sleeping. */
+    LeadContextGauge(LongSupplier clock, long ttlMillis) {
+        this.clock = clock;
+        this.ttlMillis = ttlMillis;
+    }
+
+    /** How many times this instance has actually read a transcript off disk — test seam only. */
+    int diskReadCount() {
+        return diskReads.get();
+    }
+
+    /**
+     * @param configDir the lead's {@code CLAUDE_CONFIG_DIR}, or {@code null}/blank to use the
+     *                  default {@code <user.home>/.claude} — the right answer for the common case
+     *                  where the lead's profile sets no {@code configDir} override
+     * @param sessionId the lead's own Claude session id ({@code Agent.sessionId()}), or
+     *                  {@code null} when herdr has not resolved one yet
+     * @param agentType the detected peer kind ({@code Agent.agentType()}); anything other than
+     *                  {@code "claude"} (including {@code null}, meaning undetected) reports
+     *                  {@link State#UNKNOWN} — this reader only understands Claude Code's own
+     *                  transcript format
+     */
+    public Reading read(String configDir, String sessionId, String agentType) {
+        if (sessionId == null || sessionId.isBlank()) {
+            return Reading.unknown();
+        }
+        if (!CLAUDE_AGENT_TYPE.equalsIgnoreCase(agentType)) {
+            return Reading.unknown();
+        }
+        String base = (configDir == null || configDir.isBlank())
+                ? System.getProperty("user.home") + "/.claude"
+                : configDir;
+        String cacheKey = base + '\u0000' + sessionId;
+        long now = clock.getAsLong();
+        CacheEntry cached = cache.get(cacheKey);
+        if (cached != null && now - cached.readAtMillis() < ttlMillis) {
+            return cached.reading();
+        }
+        Reading fresh = readUncached(base, sessionId);
+        cache.put(cacheKey, new CacheEntry(fresh, now));
+        return fresh;
+    }
+
+    private Reading readUncached(String base, String sessionId) {
+        diskReads.incrementAndGet();
+        Path file = findTranscript(base, sessionId);
+        if (file == null) {
+            return Reading.unknown();
+        }
+        TailRead tail;
+        try {
+            tail = tailBytes(file, TAIL_BYTES);
+        } catch (IOException e) {
+            return Reading.unknown();
+        }
+        return parse(tail);
+    }
+
+    /**
+     * Finds {@code <sessionId>.jsonl} under {@code <base>/projects/}, one level deep — never by
+     * deriving the slug directory from a working directory (see class javadoc). Bounded to a
+     * single {@code list()} of {@code projects/} itself: it never recurses further, so the cost is
+     * the number of project directories, not the size of any transcript inside them.
+     */
+    private Path findTranscript(String base, String sessionId) {
+        Path projectsDir = Path.of(base, PROJECTS_DIR);
+        if (!Files.isDirectory(projectsDir)) {
+            return null;
+        }
+        String filename = sessionId + ".jsonl";
+        Path direct = projectsDir.resolve(filename);
+        if (Files.isRegularFile(direct)) {
+            return direct;
+        }
+        try (var children = Files.list(projectsDir)) {
+            return children.filter(Files::isDirectory)
+                    .map(dir -> dir.resolve(filename))
+                    .filter(Files::isRegularFile)
+                    .findFirst()
+                    .orElse(null);
+        } catch (IOException e) {
+            return null;
+        }
+    }
+
+    /** Package-private (not {@code private}): {@link #tailBytes} is a test seam, see its javadoc. */
+    record TailRead(byte[] bytes, boolean fromStart) {
+    }
+
+    /**
+     * Reads at most {@code maxBytes} trailing bytes of {@code file}. Package-private (not
+     * {@code private}) so a test can assert directly on the returned array's length — "bytes
+     * actually read", not on any parsed answer — without needing a file anywhere near
+     * {@link #TAIL_BYTES} in size to prove the cap holds.
+     */
+    static TailRead tailBytes(Path file, int maxBytes) throws IOException {
+        try (RandomAccessFile raf = new RandomAccessFile(file.toFile(), "r")) {
+            long length = raf.length();
+            long start = Math.max(0, length - maxBytes);
+            raf.seek(start);
+            byte[] buf = new byte[(int) (length - start)];
+            raf.readFully(buf);
+            return new TailRead(buf, start == 0);
+        }
+    }
+
+    /**
+     * Parses the tail into a {@link Reading}. The first line is dropped unconditionally whenever
+     * the tail is not the whole file (it starts mid-line, cut by {@link #TAIL_BYTES} — an expected
+     * artefact of the bound, not a data problem). Every remaining line is then parsed on a
+     * best-effort basis: a line that fails to parse (most often the last one, torn by a write this
+     * read raced — see the "torn final line" section of the class javadoc) is skipped, not fatal.
+     * Only when none of the remaining lines parse does this report {@link State#UNKNOWN}.
+     */
+    private Reading parse(TailRead tail) {
+        String text = new String(tail.bytes(), StandardCharsets.UTF_8);
+        List<String> lines = new ArrayList<>(List.of(text.split("\n", -1)));
+        if (!lines.isEmpty() && lines.get(lines.size() - 1).isEmpty()) {
+            lines.remove(lines.size() - 1); // trailing newline leaves a phantom empty last element
+        }
+        if (!tail.fromStart() && !lines.isEmpty()) {
+            lines.remove(0); // first line is a fragment cut by our own tail bound, not real data
+        }
+        if (lines.isEmpty()) {
+            return Reading.unknown();
+        }
+        Long tokens = null;
+        int compactions = 0;
+        boolean anyLineParsed = false;
+        for (String line : lines) {
+            JsonNode node = tryParse(line);
+            if (node == null) {
+                // A malformed line — typically the last one, cut mid-flush by a write this read
+                // raced — is skipped rather than treated as fatal. See the class javadoc's "torn
+                // final line" section for why: a real format change makes EVERY line unparseable,
+                // not only this one, and that case is still caught below by anyLineParsed.
+                continue;
+            }
+            anyLineParsed = true;
+            JsonNode usage = node.path("message").path("usage");
+            if (usage.isObject()) {
+                tokens = usage.path("input_tokens").asLong(0)
+                        + usage.path("cache_read_input_tokens").asLong(0)
+                        + usage.path("cache_creation_input_tokens").asLong(0);
+            }
+            if ("compact_boundary".equals(node.path("subtype").asText(null))) {
+                compactions++;
+            }
+        }
+        if (!anyLineParsed) {
+            return Reading.unknown();
+        }
+        if (tokens == null) {
+            return new Reading(State.UNKNOWN, null, compactions);
+        }
+        State state = tokens >= HIGH_THRESHOLD_TOKENS ? State.HIGH : State.OK;
+        return new Reading(state, tokens, compactions);
+    }
+
+    private JsonNode tryParse(String line) {
+        try {
+            return MAPPER.readTree(line);
+        } catch (IOException e) {
+            return null;
+        }
+    }
+}

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

@@ -12,6 +12,7 @@ import dev.ltms.fleet.metrics.Metrics;
 import dev.ltms.fleet.inject.MemberPresence;
 import dev.ltms.fleet.inject.CompletionResolver;
 import dev.ltms.fleet.herdr.HerdrException;
+import dev.ltms.fleet.lead.LeadContextGauge;
 import dev.ltms.fleet.lead.LeadRollover;
 import dev.ltms.fleet.msg.LeadChannel;
 import dev.ltms.fleet.msg.LeadMessage;
@@ -115,6 +116,8 @@ public final class FleetMcp {
     private final OutageSource outage;
     /** fleetd #176: SEPARATE from both of the above — see {@link LeadSeatSource}'s doc. */
     private final LeadSeatSource leadSeats;
+    /** fleetd #602 gauge-wiring: see {@link LeadConfigDirSource}. */
+    private final LeadConfigDirSource leadConfigDirs;
     /** CB-637: this daemon's lead-to-lead channel; {@code null} when no coordinator is configured. */
     private final LeadChannel leadChannel;
     /** fleetd #361: {@code coordinator.peers} — see {@link CoordinationSource}. Empty when unset. */
@@ -126,6 +129,17 @@ public final class FleetMcp {
      * clean {@code NOT_CONFIGURED} refusal rather than throwing. See {@link #handover}.
      */
     private final LeadRollover leadRollover;
+    /**
+     * "Lead context gauge": how full each lead's own Claude Code context window is, reported on
+     * {@code fleet_list}'s {@code leads} rows (see {@link #contextView}). Built unconditionally, in
+     * the field initializer rather than a constructor parameter — this is not a togglable feature
+     * with an on/off config knob the way {@link OutageSource}/{@link LeadSeatSource} are: it needs
+     * no config at all (see {@link LeadContextGauge}'s own javadoc for the built-in defaults), so
+     * there is no "off" value to thread through every existing constructor call site. One instance
+     * per daemon so its read cache (keyed by session, TTL'd) is actually shared across
+     * {@code fleet_list} calls rather than rebuilt — and therefore useless — on every call.
+     */
+    private final LeadContextGauge leadContextGauge = new LeadContextGauge();
 
     /** Capacity facts used by {@code fleet_list}; production must supply the placement live count. */
     public record CapacitySource(Function<String, Integer> liveCount, Function<String, Integer> maxLoad,
@@ -246,6 +260,29 @@ public final class FleetMcp {
         public static LeadSeatSource none() { return new LeadSeatSource(_ -> 0); }
     }
 
+    /**
+     * fleetd #602 gauge-wiring: a lead name's configured {@code CLAUDE_CONFIG_DIR} override, fed to
+     * {@link LeadContextGauge#read} so {@code fleet_list}'s {@code context} row reads the transcript
+     * directory the lead's own profile actually writes to, not always the built-in
+     * {@code <user.home>/.claude} default.
+     *
+     * <p>The same idiom as {@link LeadSeatSource} — a {@code FleetMcp} constructor field, not a
+     * lookup {@code contextView} performs itself, because {@code FleetMcp} holds no
+     * {@link dev.ltms.fleet.config.FleetConfig} and {@code contextView} is {@code static}. See
+     * {@code Fleetd.leadConfigDirLookup} for the derivation: the same
+     * {@code fleet.leaders.<name>.profile} link {@link LeadSeatSource} already follows, resolved to
+     * that profile's own {@code configDir:}.
+     *
+     * @param configDirFor lead name → {@code configDir}, or {@code null} when the lead's entry names
+     *                      no profile, or that profile sets no {@code configDir} override — either
+     *                      way {@link LeadContextGauge} then falls back to its own built-in default,
+     *                      exactly as before this ticket
+     */
+    public record LeadConfigDirSource(Function<String, String> configDirFor) {
+        /** Inert source — every lead reads {@link LeadContextGauge}'s built-in default {@code configDir}. */
+        public static LeadConfigDirSource none() { return new LeadConfigDirSource(_ -> null); }
+    }
+
     /**
      * fleetd #361: peer-visibility facts for {@code fleet_list}'s {@code coordinator} row — this
      * daemon's own {@link LeadChannel} (for its self mailbox state and held messages) plus the
@@ -346,7 +383,7 @@ public final class FleetMcp {
                        LeadSeatSource leadSeats, List<String> peers, LeadRollover leadRollover) {
         this(messages, workers, sessions, identity, presence, primaryRegistry, callers, authorizationMode, metrics,
                 capacity, healthCoverage, LoopHealthSource.none(), quarantine, leadChannel, outage, leadSeats,
-                peers, leadRollover);
+                LeadConfigDirSource.none(), peers, leadRollover);
     }
 
     public FleetMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
@@ -354,7 +391,8 @@ public final class FleetMcp {
                     CallerResolver callers, AuthorizationMode authorizationMode, Metrics metrics,
                     CapacitySource capacity, HealthCoverageSource healthCoverage, LoopHealthSource loopHealth,
                     QuarantineSource quarantine, LeadChannel leadChannel, OutageSource outage,
-                    LeadSeatSource leadSeats, List<String> peers, LeadRollover leadRollover) {
+                    LeadSeatSource leadSeats, LeadConfigDirSource leadConfigDirs, List<String> peers,
+                    LeadRollover leadRollover) {
         Objects.requireNonNull(callers, "callers");
         this.authorizationEnforced = Objects.requireNonNull(authorizationMode, "authorizationMode")
                 == AuthorizationMode.ENFORCED;
@@ -364,6 +402,7 @@ public final class FleetMcp {
         this.quarantine = Objects.requireNonNull(quarantine, "quarantine");
         this.outage = Objects.requireNonNull(outage, "outage");
         this.leadSeats = Objects.requireNonNull(leadSeats, "leadSeats");
+        this.leadConfigDirs = Objects.requireNonNull(leadConfigDirs, "leadConfigDirs");
         this.healthCoverage = healthCoverage;
         this.loopHealth = Objects.requireNonNull(loopHealth, "loopHealth");
         this.leadRollover = leadRollover;
@@ -498,7 +537,7 @@ public final class FleetMcp {
                     McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_list", Map.of()), null);
                     if (denied != null) return denied;
                     return listFleet(workers, sessions, messages, capacity, healthCoverage, loopHealth, quarantine, outage,
-                            leadSeats, callers.leads(),
+                            leadSeats, leadContextGauge, leadConfigDirs, callers.leads(),
                             callerTerminal(exchange),
                             new CoordinationSource(leadChannel, peers),
                             coordinatorVisibleTo(principal(exchange)));
@@ -1626,7 +1665,7 @@ public final class FleetMcp {
                                                Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, loopHealth, quarantine,
-                OutageSource.none(), LeadSeatSource.none(), leads, selfTerm, coordination, false);
+                OutageSource.none(), LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1635,7 +1674,7 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, OutageSource outage,
                                                Map<String, String> leads, String selfTerm) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                LeadSeatSource.none(), leads, selfTerm, CoordinationSource.none(), false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, CoordinationSource.none(), false);
     }
 
     /**
@@ -1652,7 +1691,7 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, OutageSource.none(),
-                LeadSeatSource.none(), leads, selfTerm, coordination, false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1661,7 +1700,7 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, OutageSource outage,
                                                Map<String, String> leads, String selfTerm, CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                LeadSeatSource.none(), leads, selfTerm, coordination, false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
     }
 
     /**
@@ -1683,7 +1722,7 @@ public final class FleetMcp {
                                                LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                leadSeats, leads, selfTerm, coordination, false);
+                leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
     }
 
     /**
@@ -1707,14 +1746,24 @@ public final class FleetMcp {
                                                 LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
                                                 CoordinationSource coordination, boolean callerIsPrimary) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine,
-                outage, leadSeats, leads, selfTerm, coordination, callerIsPrimary);
+                outage, leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, callerIsPrimary);
     }
 
+    /**
+     * The canonical implementation. {@code contextGauge} is the "lead context gauge" (see
+     * {@link LeadContextGauge}) — every wrapper overload above passes a freshly constructed one,
+     * which is correct for them (none of them exercise repeated calls where a shared cache would
+     * matter); the one caller that matters for caching, {@code fleet_list}'s MCP handler, passes
+     * its own single long-lived instance instead (see {@code FleetMcp}'s {@code leadContextGauge}
+     * field).
+     */
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                                  CapacitySource capacity, HealthCoverageSource healthCoverage,
                                                  LoopHealthSource loopHealth,
                                                 QuarantineSource quarantine, OutageSource outage,
-                                               LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
+                                               LeadSeatSource leadSeats, LeadContextGauge contextGauge,
+                                               LeadConfigDirSource leadConfigDirs,
+                                               Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination, boolean callerIsPrimary) {
         try {
             Map<String, Agent> live = workers.list().stream()
@@ -1723,7 +1772,8 @@ public final class FleetMcp {
                     .collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
             List<Map<String, Object>> leadRows = leads.entrySet().stream()
                     .sorted(Map.Entry.comparingByValue())
-                    .map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm))
+                    .map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm, contextGauge,
+                            leadConfigDirs))
                     .toList();
             // fleetd #209: this is the caller-driven fleet_list read that actually reports
             // agentSessionId (via memberCapacityView -> SessionManager.rosterView), so it uses the
@@ -2018,15 +2068,25 @@ public final class FleetMcp {
     }
 
     /**
-     * One lead's row: its address, its name, and whether it can be reached right now.
+     * One lead's row: its address, its name, whether it can be reached right now, and how full its
+     * own Claude Code context window is.
      *
      * <p>{@code status} is herdr's live view, and {@code unknown} when herdr is not tracking that
      * pane as an agent — the honest answer, and the one that matters: a lead whose pane herdr cannot
      * see is a lead a {@code fleet_send} cannot be typed into. It is reported rather than hidden,
      * because a peer that has gone unreachable is exactly what the sender needs to know.
+     *
+     * <p>{@code context} is the "lead context gauge" (fleetd's context-usage visibility ticket):
+     * {@code {state: "ok"|"high"|"unknown", tokens?: number, compactions: number}}, read from the
+     * lead's transcript — see {@link LeadContextGauge}. Kept small on purpose (a token count, a
+     * state, and a compaction count) rather than echoing the whole reading history: this is a
+     * roster row a caller glances at, not a diagnostics dump. {@code tokens} is present only when
+     * {@code state} is not {@code "unknown"} — never a stale or default number standing in for "I
+     * could not tell".
      */
     private static Map<String, Object> leadView(String terminal, String name, Agent live,
-                                                String selfTerm) {
+                                                String selfTerm, LeadContextGauge contextGauge,
+                                                LeadConfigDirSource leadConfigDirs) {
         Map<String, Object> m = new LinkedHashMap<>();
         m.put("sessionId", terminal);
         m.put("name", name);
@@ -2035,9 +2095,32 @@ public final class FleetMcp {
         if (terminal.equals(selfTerm)) {
             m.put("self", true);
         }
+        String configDir = leadConfigDirs.configDirFor().apply(name);
+        m.put("context", contextView(contextGauge, live, configDir));
         return m;
     }
 
+    /**
+     * Reads the lead context gauge for one lead. {@code configDir} is this lead's configured
+     * {@code CLAUDE_CONFIG_DIR} override (see {@link LeadConfigDirSource}), derived from
+     * {@code fleet.leaders.<name>.profile} → that profile's own {@code configDir:} — or {@code null}
+     * when the lead's entry names no profile, or that profile sets no override, in which case
+     * {@link LeadContextGauge#read} falls back to its own built-in default
+     * ({@code <user.home>/.claude}).
+     */
+    private static Map<String, Object> contextView(LeadContextGauge contextGauge, Agent live, String configDir) {
+        String sessionId = live == null ? null : live.sessionId();
+        String agentType = live == null ? null : live.agentType();
+        LeadContextGauge.Reading reading = contextGauge.read(configDir, sessionId, agentType);
+        Map<String, Object> c = new LinkedHashMap<>();
+        c.put("state", reading.state().name().toLowerCase());
+        if (reading.tokens() != null) {
+            c.put("tokens", reading.tokens());
+        }
+        c.put("compactions", reading.compactions());
+        return c;
+    }
+
     /** {@code fleet_stop}: tear a worker down by its pane id. */
     static McpSchema.CallToolResult stop(SessionManager sessions, String paneId) {
         if (isBlank(paneId)) {

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

@@ -0,0 +1,100 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.FleetConfig;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * fleetd #602 gauge-wiring: {@link Fleetd#leadConfigDirLookup} is the factory {@code Fleetd.main}
+ * wires into {@code FleetMcp.LeadConfigDirSource} so {@code fleet_list}'s {@code context} row reads
+ * the transcript directory a lead's OWN profile actually writes to, instead of always falling back
+ * to the built-in {@code <user.home>/.claude} default (see {@code LeadContextGauge}).
+ *
+ * <p>{@code FleetMcpLeadContextGaugeWiringTest} proves the directory this factory returns is what
+ * actually gets read; this class proves the factory's own matching logic — the same
+ * {@code fleet.leaders.<name>.profile} link {@code Fleetd.leadSeatLookup} already follows (see
+ * {@code FleetdLeadSeatLookupTest}), one step further to that profile's own {@code configDir:}.
+ */
+class FleetdLeadConfigDirLookupTest {
+
+    private static FleetConfig.Profile profileWithConfigDir(String name, String configDir) {
+        return new FleetConfig.Profile(name, null, "claude-sonnet-5", configDir, null, null,
+                "tab", "fleet", "w #{n}", null, null, null, null, null, null, null,
+                null, 3, true, null, null, null);
+    }
+
+    private static FleetConfig.Leader leadOnProfile(String profile) {
+        return new FleetConfig.Leader(profile, "lead: primary", 1, "lead:", 10, "claude", "claude-sonnet-5");
+    }
+
+    @Test
+    @DisplayName("a lead on a profile that sets configDir resolves to that directory")
+    void leadOnAProfileWithConfigDirResolvesToIt() {
+        Map<String, FleetConfig.Profile> profiles = Map.of("opus", profileWithConfigDir("opus", "/mnt/opus-claude"));
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("opus"));
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(() -> profiles, leaders);
+
+        assertEquals("/mnt/opus-claude", lookup.apply("primary"));
+    }
+
+    @Test
+    @DisplayName("changing the config from directory A to directory B changes what the lookup reports")
+    void configChangeFromDirectoryAToDirectoryBChangesTheAnswer() {
+        java.util.concurrent.atomic.AtomicReference<Map<String, FleetConfig.Profile>> profilesRef =
+                new java.util.concurrent.atomic.AtomicReference<>(
+                        Map.of("opus", profileWithConfigDir("opus", "/mnt/dir-a")));
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("opus"));
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(profilesRef::get, leaders);
+
+        assertEquals("/mnt/dir-a", lookup.apply("primary"), "must read directory A before the config changes");
+
+        profilesRef.set(Map.of("opus", profileWithConfigDir("opus", "/mnt/dir-b")));
+        assertEquals("/mnt/dir-b", lookup.apply("primary"), "must read directory B once the LIVE config changes — "
+                + "a lookup that snapshotted the profile map at construction would still answer directory A here");
+    }
+
+    @Test
+    @DisplayName("a lead entry with no `profile:` (recognise-only) resolves to null, not a thrown exception")
+    void recogniseOnlyLeadWithNoProfileResolvesToNull() {
+        Map<String, FleetConfig.Profile> profiles = Map.of("opus", profileWithConfigDir("opus", "/mnt/opus-claude"));
+        FleetConfig.Leader recogniseOnly = new FleetConfig.Leader(null, "lead: primary", 1, "lead:", 10,
+                "claude", "claude-sonnet-5");
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", recogniseOnly);
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(() -> profiles, leaders);
+
+        assertNull(lookup.apply("primary"));
+    }
+
+    @Test
+    @DisplayName("a lead naming a profile that is not configured resolves to null, not a thrown exception")
+    void leadOnAnUnconfiguredProfileResolvesToNull() {
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("ghost-profile"));
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(Map::of, leaders);
+
+        assertNull(lookup.apply("primary"));
+    }
+
+    @Test
+    @DisplayName("a lead on a profile that sets no configDir override resolves to null")
+    void leadOnAProfileWithNoConfigDirResolvesToNull() {
+        Map<String, FleetConfig.Profile> profiles = Map.of("opus", profileWithConfigDir("opus", null));
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("opus"));
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(() -> profiles, leaders);
+
+        assertNull(lookup.apply("primary"));
+    }
+
+    @Test
+    @DisplayName("an unrecognised lead name resolves to null, not a thrown exception")
+    void unrecognisedLeadNameResolvesToNull() {
+        Function<String, String> lookup = Fleetd.leadConfigDirLookup(Map::of, Map.of());
+
+        assertNull(lookup.apply("ghost-lead"));
+    }
+}

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

@@ -0,0 +1,98 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.mcp.FleetMcp;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * fleetd #602 gauge-wiring follow-up (PR #606 review comment 17353): {@code Fleetd.main}'s {@code
+ * LeadConfigDirSource} local used to be a bare {@code new FleetMcp.LeadConfigDirSource(
+ * leadConfigDirLookup(...))} built inline, with nothing a test could call directly. Measured on
+ * that shape: replacing the whole expression with {@code FleetMcp.LeadConfigDirSource.none()} at
+ * the call site compiled with 0 errors and left the full 1822-test suite green — the daemon could
+ * be changed to always report every lead's context as {@code UNKNOWN}, forever, and no test would
+ * notice. That is the same hand-built-vs-config-wired shape as fleetd #561/#248/#426/#562
+ * ({@code FleetdLoopHealthSourceWiringTest}).
+ *
+ * <p>{@code FleetMcpLeadContextGaugeWiringTest} and {@code FleetdLeadConfigDirLookupTest} both
+ * predate this class and are both still correct — but neither can catch the mutation above. One
+ * builds its own {@code FleetMcp} and hands it its own {@code LeadConfigDirSource}; the other
+ * builds its own lookup and calls {@link Fleetd#leadConfigDirLookup} directly. Neither one ever
+ * calls the thing {@code Fleetd.main} actually calls.
+ *
+ * <p>The fix extracts the inline {@code new} into {@link Fleetd#leadConfigDirSource}, a
+ * package-private factory in the same style as {@link Fleetd#loopHealthSource}/{@link
+ * Fleetd#capacitySource}/{@link Fleetd#healthCoverageSource} — which is exactly what makes it
+ * directly callable here. This test calls that factory with real {@link FleetConfig.Profile}/
+ * {@link FleetConfig.Leader} fixtures (the same shapes {@code FleetdLeadConfigDirLookupTest}
+ * already uses) and asserts the returned source resolves a real {@code configDir} — a property
+ * that would be false if {@link Fleetd#leadConfigDirSource} were mutated to {@code return
+ * FleetMcp.LeadConfigDirSource.none();}. Measured: mutating exactly that line makes
+ * {@link #resolvesTheRealConfiguredConfigDir()} fail ({@code expected: </mnt/opus-claude> but was:
+ * <null>}); restoring it makes the whole suite green again.
+ *
+ * <p><b>What this class does not and cannot cover.</b> {@code main}'s own line —
+ * {@code leadConfigDirSource(() -> config.get().profiles(), leaders)} — could itself be swapped
+ * for a bare {@code FleetMcp.LeadConfigDirSource.none()}, bypassing this factory entirely. Measured:
+ * that exact mutation compiles with 0 errors and leaves every test in this file, and the full
+ * 1825-test suite, green. {@link Fleetd#loopHealthSource}'s own wiring test has the identical gap
+ * for its own one-line call in {@code main} — no test in this codebase calls {@code Fleetd.main}
+ * far enough to observe which factory call it made. This class narrows the gap from "nothing tests
+ * the wiring" (the pre-extraction state this ticket found) to "the factory's own logic is pinned,
+ * and main's call to it is a one-line, visually-verifiable delegation" — the same standard already
+ * accepted for {@code loopHealthSource}/{@code capacitySource}/{@code healthCoverageSource}.
+ */
+class FleetdLeadConfigDirSourceWiringTest {
+
+    private static FleetConfig.Profile profileWithConfigDir(String name, String configDir) {
+        return new FleetConfig.Profile(name, null, "claude-sonnet-5", configDir, null, null,
+                "tab", "fleet", "w #{n}", null, null, null, null, null, null, null,
+                null, 3, true, null, null, null);
+    }
+
+    private static FleetConfig.Leader leadOnProfile(String profile) {
+        return new FleetConfig.Leader(profile, "lead: primary", 1, "lead:", 10, "claude", "claude-sonnet-5");
+    }
+
+    @Test
+    @DisplayName("the returned source resolves the lead's REAL configured configDir, not a hardcoded null")
+    void resolvesTheRealConfiguredConfigDir() {
+        Map<String, FleetConfig.Profile> profiles = Map.of("opus", profileWithConfigDir("opus", "/mnt/opus-claude"));
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("opus"));
+
+        FleetMcp.LeadConfigDirSource source = Fleetd.leadConfigDirSource(() -> profiles, leaders);
+
+        assertEquals("/mnt/opus-claude", source.configDirFor().apply("primary"),
+                "the configDirFor function must delegate to the real leadConfigDirLookup — mutating "
+                        + "Fleetd.leadConfigDirSource's own body to `return FleetMcp.LeadConfigDirSource.none();` "
+                        + "must fail this assertion (measured: it does — see this test's class javadoc for the "
+                        + "companion measurement on main's one-line call to this factory, which this assertion "
+                        + "does not and structurally cannot cover)");
+    }
+
+    @Test
+    @DisplayName("a lead on a profile with no configDir override still resolves to null, not a crash")
+    void leadWithNoConfigDirOverrideResolvesToNull() {
+        Map<String, FleetConfig.Profile> profiles = Map.of("opus", profileWithConfigDir("opus", null));
+        Map<String, FleetConfig.Leader> leaders = Map.of("primary", leadOnProfile("opus"));
+
+        FleetMcp.LeadConfigDirSource source = Fleetd.leadConfigDirSource(() -> profiles, leaders);
+
+        assertNull(source.configDirFor().apply("primary"),
+                "no configDir: override configured ⇒ null, so LeadContextGauge falls back to its own default");
+    }
+
+    @Test
+    @DisplayName("an unrecognised lead name resolves to null, not a thrown exception")
+    void unrecognisedLeadNameResolvesToNull() {
+        FleetMcp.LeadConfigDirSource source = Fleetd.leadConfigDirSource(Map::of, Map.of());
+
+        assertNull(source.configDirFor().apply("ghost-lead"));
+    }
+}

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

@@ -0,0 +1,238 @@
+package dev.ltms.fleet.lead;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Ticket "lead context gauge" — fleetd could not see how full a lead's own Claude Code context
+ * window was, so a lead auto-compacting was always a surprise. {@link LeadContextGauge} reads that
+ * off the lead's own transcript. Five acceptance properties from the ticket, one test class each
+ * (plus the three separate UNKNOWN cases the ticket calls out by name):
+ * <ol>
+ *   <li>{@link #tokenCountTracksTheLastUsageRecordAndChangesWithIt()}</li>
+ *   <li>{@link #compactionCountTracksCompactBoundaryRecordsAndChangesWithIt()}</li>
+ *   <li>{@link #missingFileIsUnknown()}, {@link #unreadableFileIsUnknown()}</li>
+ *   <li>{@link #tornFinalLineFallsBackToTheLastGoodReading()},
+ *       {@link #everyLineUnparseableIsUnknown()} (fleetd #602 gauge-wiring, finding 2)</li>
+ *   <li>{@link #readNeverExceedsTheTailBound()}</li>
+ *   <li>{@link #secondReadInsideTtlDoesNotTouchDiskAgain()}</li>
+ * </ol>
+ *
+ * <p>Every fixture lives under {@code @TempDir} — never the operator's real config directory (see
+ * the ticket's hard constraint on this).
+ *
+ * <p><strong>fleetd #602 gauge-wiring, finding 2.</strong> The property above used to be "a
+ * truncated or invalid LAST line reports UNKNOWN" — on the theory that a bad last line signals a
+ * format change. That reasoning did not hold: {@code fleet_list} reads this transcript while Claude
+ * Code may be mid-write on it, so a torn LAST line is an ordinary race, not a format change, and a
+ * real format change makes EVERY line unparseable, not only the last one written. So the property
+ * is now split in two: {@link #tornFinalLineFallsBackToTheLastGoodReading} (the torn-line case must
+ * NOT destroy a good earlier reading) and its control, {@link #everyLineUnparseableIsUnknown} (only
+ * when NOTHING in the window parses does this report UNKNOWN).
+ */
+class LeadContextGaugeTest {
+
+    private static final String SESSION_ID = "11111111-1111-1111-1111-111111111111";
+
+    /** One line Claude Code would write for a turn with the given live-context total. */
+    private static String usageLine(long inputTokens, long cacheRead, long cacheCreation) {
+        return "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"usage\":{"
+                + "\"input_tokens\":" + inputTokens + ","
+                + "\"cache_read_input_tokens\":" + cacheRead + ","
+                + "\"cache_creation_input_tokens\":" + cacheCreation + "}}}";
+    }
+
+    /** One line Claude Code writes when an auto-compaction happens. */
+    private static String compactionLine() {
+        return "{\"type\":\"system\",\"subtype\":\"compact_boundary\","
+                + "\"compactMetadata\":{\"preTokens\":250000,\"postTokens\":5000,"
+                + "\"trigger\":\"auto\",\"durationMs\":54000}}";
+    }
+
+    /** Lays out {@code <configDir>/projects/<anySlug>/<sessionId>.jsonl} and writes {@code lines}. */
+    private static String writeTranscript(Path configDir, String sessionId, String... lines) throws IOException {
+        Path projectDir = configDir.resolve("projects").resolve("some-project-slug");
+        Files.createDirectories(projectDir);
+        Path file = projectDir.resolve(sessionId + ".jsonl");
+        StringBuilder sb = new StringBuilder();
+        for (String line : lines) {
+            sb.append(line).append('\n');
+        }
+        Files.writeString(file, sb.toString(), StandardCharsets.UTF_8);
+        return configDir.toString();
+    }
+
+    // --- property 1: live token count tracks the last usage record -----------------------------
+
+    @Test
+    @DisplayName("reported tokens equal the LAST usage record's total, and change when it does")
+    void tokenCountTracksTheLastUsageRecordAndChangesWithIt(@TempDir Path tmp) throws IOException {
+        String configDir = writeTranscript(tmp, SESSION_ID,
+                usageLine(1_000, 0, 0),
+                usageLine(40_000, 5_000, 3_000)); // last record: 48,000
+        LeadContextGauge gauge = new LeadContextGauge();
+
+        LeadContextGauge.Reading first = gauge.read(configDir, SESSION_ID, "claude");
+        assertEquals(48_000L, first.tokens(), "must total input+cache_read+cache_creation of the LAST usage record");
+        assertEquals(LeadContextGauge.State.OK, first.state());
+
+        // Change N in the fixture (a fresh session id avoids the cache) -- the reported number
+        // must change with it, not stay pinned to the first fixture's total.
+        String otherSession = "22222222-2222-2222-2222-222222222222";
+        writeTranscript(tmp, otherSession, usageLine(100_000, 50_000, 50_000)); // last record: 200,000
+        LeadContextGauge.Reading second = gauge.read(configDir, otherSession, "claude");
+        assertEquals(200_000L, second.tokens());
+        assertTrue(second.tokens() != first.tokens(), "changing N in the fixture must change the reported number");
+    }
+
+    // --- property 2: compaction count tracks compact_boundary records --------------------------
+
+    @Test
+    @DisplayName("reported compaction count equals K compact_boundary records, and changes with K")
+    void compactionCountTracksCompactBoundaryRecordsAndChangesWithIt(@TempDir Path tmp) throws IOException {
+        String sessionTwoCompactions = "33333333-3333-3333-3333-333333333333";
+        writeTranscript(tmp, sessionTwoCompactions,
+                usageLine(1_000, 0, 0),
+                compactionLine(),
+                usageLine(2_000, 0, 0),
+                compactionLine(),
+                usageLine(3_000, 0, 0));
+        LeadContextGauge gauge = new LeadContextGauge();
+        LeadContextGauge.Reading twoCompactions = gauge.read(tmp.toString(), sessionTwoCompactions, "claude");
+        assertEquals(2, twoCompactions.compactions());
+
+        String sessionZeroCompactions = "44444444-4444-4444-4444-444444444444";
+        writeTranscript(tmp, sessionZeroCompactions, usageLine(3_000, 0, 0));
+        LeadContextGauge.Reading zeroCompactions = gauge.read(tmp.toString(), sessionZeroCompactions, "claude");
+        assertEquals(0, zeroCompactions.compactions(), "changing K in the fixture must change the reported count");
+    }
+
+    // --- property 3: two separate UNKNOWN cases -------------------------------------------------
+
+    @Test
+    @DisplayName("a missing transcript file reports UNKNOWN with no token number")
+    void missingFileIsUnknown(@TempDir Path tmp) {
+        LeadContextGauge gauge = new LeadContextGauge();
+        LeadContextGauge.Reading reading = gauge.read(tmp.toString(), SESSION_ID, "claude");
+        assertEquals(LeadContextGauge.State.UNKNOWN, reading.state());
+        assertNull(reading.tokens());
+    }
+
+    @Test
+    @DisplayName("an unreadable transcript file reports UNKNOWN with no token number")
+    void unreadableFileIsUnknown(@TempDir Path tmp) throws IOException {
+        String configDir = writeTranscript(tmp, SESSION_ID, usageLine(1_000, 0, 0));
+        Path file = tmp.resolve("projects").resolve("some-project-slug").resolve(SESSION_ID + ".jsonl");
+        assertTrue(file.toFile().setReadable(false), "test setup: must be able to revoke read permission");
+        try {
+            LeadContextGauge gauge = new LeadContextGauge();
+            LeadContextGauge.Reading reading = gauge.read(configDir, SESSION_ID, "claude");
+            assertEquals(LeadContextGauge.State.UNKNOWN, reading.state());
+            assertNull(reading.tokens());
+        } finally {
+            file.toFile().setReadable(true); // so @TempDir cleanup can delete it
+        }
+    }
+
+    @Test
+    @DisplayName("fleetd #602 finding 2: a torn final line does not destroy a good earlier reading")
+    void tornFinalLineFallsBackToTheLastGoodReading(@TempDir Path tmp) throws IOException {
+        // Deliberately NOT via writeTranscript: that helper appends a trailing newline after every
+        // line, including the last one, which would not model what a write caught mid-flush looks
+        // like on disk. Claude Code appends and flushes one line at a time, so the torn line here has
+        // no trailing newline at all -- exactly the shape a read racing an in-progress write sees.
+        Path projectDir = tmp.resolve("projects").resolve("some-project-slug");
+        Files.createDirectories(projectDir);
+        Path file = projectDir.resolve(SESSION_ID + ".jsonl");
+        String lastCompleteLine = usageLine(1_000, 2_000, 3_000); // last COMPLETE record: 6,000
+        String tornLine = "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"usage\":{\"input_tok";
+        Files.writeString(file, lastCompleteLine + "\n" + tornLine, StandardCharsets.UTF_8);
+
+        LeadContextGauge gauge = new LeadContextGauge();
+        LeadContextGauge.Reading reading = gauge.read(tmp.toString(), SESSION_ID, "claude");
+        assertEquals(LeadContextGauge.State.OK, reading.state(),
+                "a torn final line must not turn a good earlier reading into UNKNOWN");
+        assertEquals(6_000L, reading.tokens(),
+                "must report the last COMPLETE line's total, not fail the whole read");
+    }
+
+    @Test
+    @DisplayName("fleetd #602 finding 2 control: when EVERY line is unparseable, the reading IS UNKNOWN")
+    void everyLineUnparseableIsUnknown(@TempDir Path tmp) throws IOException {
+        // The real signal a format change gives: not one bad line, but ALL of them. Without this
+        // control, code that never reports UNKNOWN at all would still pass the torn-line property.
+        String configDir = writeTranscript(tmp, SESSION_ID,
+                "{this is not json at all",
+                "neither is this{{{");
+        LeadContextGauge gauge = new LeadContextGauge();
+        LeadContextGauge.Reading reading = gauge.read(configDir, SESSION_ID, "claude");
+        assertEquals(LeadContextGauge.State.UNKNOWN, reading.state(),
+                "every line unparseable is the real format-change signal and must still report UNKNOWN");
+        assertNull(reading.tokens());
+    }
+
+    // --- property 4: the tail bound is actually enforced ----------------------------------------
+
+    @Test
+    @DisplayName("reading a transcript far larger than the tail bound reads no more than the bound")
+    void readNeverExceedsTheTailBound(@TempDir Path tmp) throws IOException {
+        int smallBound = 4_096; // exercise the mechanism without writing a multi-MB fixture
+        Path file = tmp.resolve("big.jsonl");
+        // One line far bigger than smallBound, repeated, so the whole file is many times the bound.
+        String line = usageLine(42, 0, 0) + " ".repeat(500);
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < 50; i++) {
+            sb.append(line).append('\n');
+        }
+        Files.writeString(file, sb.toString(), StandardCharsets.UTF_8);
+        long fileLength = Files.size(file);
+        assertTrue(fileLength > (long) smallBound * 5, "test setup: fixture must genuinely dwarf the bound");
+
+        var tail = LeadContextGauge.tailBytes(file, smallBound);
+        assertEquals(smallBound, tail.bytes().length,
+                "must read exactly the bound, not the whole " + fileLength + "-byte file — assert on bytes actually read");
+    }
+
+    // --- property 5: the cache TTL means a burst of calls reads the file once -------------------
+
+    @Test
+    @DisplayName("the same (configDir, sessionId) read twice inside the TTL touches disk once")
+    void secondReadInsideTtlDoesNotTouchDiskAgain(@TempDir Path tmp) throws IOException {
+        String configDir = writeTranscript(tmp, SESSION_ID, usageLine(1_000, 0, 0));
+        AtomicLong now = new AtomicLong(0);
+        LeadContextGauge gauge = new LeadContextGauge(now::get, 5_000);
+
+        gauge.read(configDir, SESSION_ID, "claude");
+        gauge.read(configDir, SESSION_ID, "claude"); // still inside the TTL window
+        assertEquals(1, gauge.diskReadCount(), "two reads inside the TTL must touch disk once");
+
+        now.set(6_000); // past the TTL
+        gauge.read(configDir, SESSION_ID, "claude");
+        assertEquals(2, gauge.diskReadCount(), "a read past the TTL must touch disk again");
+    }
+
+    // --- non-Claude peer / unresolved session -----------------------------------------------------
+
+    @Test
+    @DisplayName("a non-claude agent type or a null session id is UNKNOWN, never OK")
+    void nonClaudePeerOrUnresolvedSessionIsUnknown(@TempDir Path tmp) throws IOException {
+        String configDir = writeTranscript(tmp, SESSION_ID, usageLine(1_000, 0, 0));
+        LeadContextGauge gauge = new LeadContextGauge();
+
+        assertEquals(LeadContextGauge.State.UNKNOWN, gauge.read(configDir, SESSION_ID, "opencode").state());
+        assertEquals(LeadContextGauge.State.UNKNOWN, gauge.read(configDir, SESSION_ID, null).state());
+        assertEquals(LeadContextGauge.State.UNKNOWN, gauge.read(configDir, null, "claude").state());
+    }
+}

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

@@ -0,0 +1,126 @@
+package dev.ltms.fleet.mcp;
+
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.WorkspaceControl;
+import dev.ltms.fleet.lead.LeadContextGauge;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
+import dev.ltms.fleet.session.SessionManager;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #602 gauge-wiring: {@code FleetMcp}'s {@code fleet_list} handler shipped calling
+ * {@code LeadContextGauge.read(null, ...)} unconditionally, so every lead whose profile sets a
+ * {@code configDir:} override read the wrong transcript directory forever, with no error anywhere —
+ * {@link LeadContextGauge}'s own 8 tests all passed because none of them exercised THIS wiring; each
+ * one hands the gauge its own directory directly.
+ *
+ * <p>This class proves the directory {@code fleet.leaders.<name>.profile}'s own {@code configDir:}
+ * names is the one {@code fleet_list}'s {@code context} row actually reads — not merely that SOME
+ * directory got passed. If the wiring in {@code FleetMcp.leadView}/{@code contextView} is ever
+ * reverted to a hardcoded {@code null}, {@link #configuredDirectoryDecidesWhichTranscriptIsRead()}
+ * must go red: both directories below hold a REAL, DIFFERENT token count for the SAME session id,
+ * so a hardcoded {@code null} (always reading the built-in default, where neither transcript lives)
+ * would report {@code UNKNOWN} both times instead of the two distinct numbers this test asserts.
+ */
+class FleetMcpLeadContextGaugeWiringTest {
+
+    private static final String LEAD_TERMINAL = "term_lead";
+    private static final String LEAD_NAME = "opus";
+    /** {@code FakeHerdr.withAgent} always projects {@code agent_session.value} as {@code sess-<terminalId>}. */
+    private static final String LEAD_SESSION_ID = "sess-" + LEAD_TERMINAL;
+
+    private static ClaudeCodeLauncher workerService(FakeHerdr h) {
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        return new ClaudeCodeLauncher(new AgentControl(h), new WorkspaceControl(h),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> "tok");
+    }
+
+    private static String textOf(McpSchema.CallToolResult r) {
+        return ((McpSchema.TextContent) r.content().getFirst()).text();
+    }
+
+    /** One line Claude Code would write for a turn with the given live-context total. */
+    private static String usageLine(long tokens) {
+        return "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"usage\":{"
+                + "\"input_tokens\":" + tokens + ",\"cache_read_input_tokens\":0,\"cache_creation_input_tokens\":0}}}";
+    }
+
+    /** Lays out {@code <configDir>/projects/<anySlug>/<sessionId>.jsonl} carrying one usage record. */
+    private static String writeTranscript(Path configDir, String sessionId, long tokens) throws IOException {
+        Path projectDir = configDir.resolve("projects").resolve("some-project-slug");
+        Files.createDirectories(projectDir);
+        Files.writeString(projectDir.resolve(sessionId + ".jsonl"), usageLine(tokens) + "\n", StandardCharsets.UTF_8);
+        return configDir.toString();
+    }
+
+    @Test
+    @DisplayName("the CONFIGURED directory decides which transcript is read, and follows a config change")
+    void configuredDirectoryDecidesWhichTranscriptIsRead(@TempDir Path tmp) throws IOException {
+        Path dirA = Files.createDirectory(tmp.resolve("dir-a"));
+        Path dirB = Files.createDirectory(tmp.resolve("dir-b"));
+        writeTranscript(dirA, LEAD_SESSION_ID, 11_000);
+        writeTranscript(dirB, LEAD_SESSION_ID, 22_000);
+
+        FakeHerdr herdr = new FakeHerdr().withAgent("lead-opus", LEAD_TERMINAL, "wL:p1", "wL:t1");
+        SessionManager sessions = new SessionManager(workerService(herdr));
+        LeadContextGauge contextGauge = new LeadContextGauge();
+        AtomicReference<String> configuredDir = new AtomicReference<>(dirA.toString());
+        FleetMcp.LeadConfigDirSource source = new FleetMcp.LeadConfigDirSource(name ->
+                LEAD_NAME.equals(name) ? configuredDir.get() : null);
+
+        String firstRead = textOf(listFleet(herdr, sessions, contextGauge, source));
+        assertTrue(firstRead.contains("\"tokens\":11000"),
+                "config names directory A ⇒ fleet_list must report A's token count: " + firstRead);
+
+        // The config changes to name directory B instead -- the very next read must follow it.
+        configuredDir.set(dirB.toString());
+        String secondRead = textOf(listFleet(herdr, sessions, contextGauge, source));
+        assertTrue(secondRead.contains("\"tokens\":22000"),
+                "config now names directory B ⇒ fleet_list must report B's token count, not A's stale one: "
+                        + secondRead);
+    }
+
+    @Test
+    @DisplayName("a lead whose config names no directory still degrades to the built-in default, never throws")
+    void aLeadWhoseConfigNamesNoDirectoryStillFallsBackWithoutThrowing() {
+        FakeHerdr herdr = new FakeHerdr().withAgent("lead-opus", LEAD_TERMINAL, "wL:p1", "wL:t1");
+        SessionManager sessions = new SessionManager(workerService(herdr));
+        LeadContextGauge contextGauge = new LeadContextGauge();
+
+        McpSchema.CallToolResult res = listFleet(herdr, sessions, contextGauge, FleetMcp.LeadConfigDirSource.none());
+
+        assertNotEquals(Boolean.TRUE, res.isError(), "a lead with no configured configDir must degrade, never throw");
+        String out = textOf(res);
+        assertTrue(out.contains("\"state\":\"unknown\""), "no transcript under the built-in default ⇒ UNKNOWN: " + out);
+        assertFalse(out.contains("\"tokens\""), "UNKNOWN must never carry a stale/default token number: " + out);
+    }
+
+    private static McpSchema.CallToolResult listFleet(FakeHerdr herdr, SessionManager sessions,
+            LeadContextGauge contextGauge, FleetMcp.LeadConfigDirSource leadConfigDirs) {
+        return FleetMcp.listFleet(workerService(herdr), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.LoopHealthSource.none(), FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(),
+                FleetMcp.LeadSeatSource.none(), contextGauge, leadConfigDirs,
+                Map.of(LEAD_TERMINAL, LEAD_NAME), LEAD_TERMINAL, FleetMcp.CoordinationSource.none(), false);
+    }
+}

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

@@ -0,0 +1,168 @@
+package dev.ltms.fleet.msg;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A {@link ScheduledExecutorService} for tests that never runs a task on a timer — it records what
+ * {@link ReplyPushLoop} schedules and only ever runs it when the test itself calls
+ * {@link #runDueTasks()} (fleetd #608).
+ *
+ * <p>The test this exists for ({@code anAlreadyCollectedTicketProducesNoNudge}) used to wire
+ * {@link ReplyPushLoop} to a real {@code Executors.newSingleThreadScheduledExecutor()} and bet a
+ * 300ms backoff was "wide" enough that the test's own work (collecting the ticket) always won the
+ * race against the scheduler's own timer firing the next tick. On an idle machine that held; under
+ * a loaded full-suite run it did not, and the test went red on perfectly correct code. Replacing
+ * the timer with this fake removes the race rather than widening it: nothing here ever runs on a
+ * schedule of its own, so no backoff value — however small — can make the tick fire before the test
+ * is ready for it.
+ *
+ * <p><strong>Deliberately narrow.</strong> {@link ReplyPushLoop} calls exactly two methods on its
+ * {@link ScheduledExecutorService} field — {@link #schedule(Runnable, long, TimeUnit)} (every tick,
+ * including the very first one {@code startOrCoalesce} kicks off) and {@link #shutdownNow()} (on
+ * {@code ReplyPushLoop.stop()}/{@code close()}) — verified by reading every {@code scheduler.} call
+ * site in that class. Every other {@link ScheduledExecutorService} method throws
+ * {@link UnsupportedOperationException} rather than silently doing the wrong thing, so a future
+ * change to {@code ReplyPushLoop} that starts calling one of them fails this fake loudly, on the
+ * very first test that exercises it, instead of being quietly mishandled.
+ */
+final class ManualScheduler implements ScheduledExecutorService {
+
+    private final Deque<Runnable> pending = new ArrayDeque<>();
+    private boolean shutdown = false;
+
+    /**
+     * Run every task pending as of the START of this call — a snapshot taken before anything runs.
+     * {@link ReplyPushLoop#tick} always reschedules its own next tick before returning (see
+     * {@code scheduleNext} in its {@code INJECT}/{@code WAIT_BUSY} branches), so without the
+     * snapshot a single call here would recurse forever. Taking it up front means one call is
+     * exactly one tick, deterministically, no matter what that tick itself goes on to schedule.
+     *
+     * @return how many tasks actually ran
+     */
+    synchronized int runDueTasks() {
+        List<Runnable> due = new ArrayList<>(pending);
+        pending.clear();
+        for (Runnable task : due) {
+            task.run();
+        }
+        return due.size();
+    }
+
+    /** How many tasks are currently queued, without running any of them. */
+    synchronized int pendingCount() {
+        return pending.size();
+    }
+
+    @Override
+    public synchronized ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        if (shutdown) {
+            throw new RejectedExecutionException("ManualScheduler is shut down");
+        }
+        pending.add(command);
+        return null; // ReplyPushLoop discards the return value of every schedule() call it makes.
+    }
+
+    @Override
+    public synchronized List<Runnable> shutdownNow() {
+        shutdown = true;
+        List<Runnable> left = new ArrayList<>(pending);
+        pending.clear();
+        return left;
+    }
+
+    @Override
+    public synchronized boolean isShutdown() {
+        return shutdown;
+    }
+
+    // --- everything below: not called by ReplyPushLoop, and not supported by this fake (see the
+    // class javadoc's "deliberately narrow" note) -------------------------------------------------
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+        throw unsupported();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        throw unsupported();
+    }
+
+    @Override
+    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+        throw unsupported();
+    }
+
+    @Override
+    public void shutdown() {
+        throw unsupported();
+    }
+
+    @Override
+    public boolean isTerminated() {
+        throw unsupported();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        throw unsupported();
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws ExecutionException {
+        throw unsupported();
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+            throws ExecutionException {
+        throw unsupported();
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        throw unsupported();
+    }
+
+    private static UnsupportedOperationException unsupported() {
+        return new UnsupportedOperationException(
+                "ManualScheduler only supports schedule(Runnable, long, TimeUnit), shutdownNow() and isShutdown() "
+                        + "— see the class javadoc");
+    }
+}

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

@@ -1841,6 +1841,36 @@ class MessageServiceTest {
         return new PushWiring(service, leadHerdr, scheduler);
     }
 
+    /**
+     * A {@link MessageService} wired to a real {@link ReplyPushLoop}, like {@link #wireWithPushLoop},
+     * but backed by {@link ManualScheduler} instead of a real timer (fleetd #608): its tick never
+     * fires on its own — a test drives it explicitly via {@link ManualScheduler#runDueTasks()}.
+     */
+    private record ManualPushWiring(MessageService service, FakeHerdr leadHerdr, ManualScheduler scheduler)
+            implements AutoCloseable {
+        @Override
+        public void close() {
+            scheduler.shutdownNow();
+        }
+    }
+
+    /**
+     * As {@link #wireWithPushLoop}, but the schedule never runs on its own. {@code backoffMs} is
+     * still threaded through to {@link ReplyPushLoop}'s constructor (it takes one), but nothing here
+     * ever waits it out, so its value cannot affect anything a test built on this observes — see
+     * {@code anAlreadyCollectedTicketProducesNoNudge}, which sets it to 1 to prove exactly that.
+     */
+    private ManualPushWiring wireWithManualScheduler(int maxReminders, long backoffMs) {
+        PrimaryRegistry registry = new PrimaryRegistry(null);
+        registry.recordDelegation(T, LEAD);
+        FakeHerdr leadHerdr = new FakeHerdr();
+        AgentControl leadAgents = new AgentControl(leadHerdr);
+        ManualScheduler scheduler = new ManualScheduler();
+        ReplyPushLoop pushLoop = new ReplyPushLoop(registry, leadAgents, inbox, scheduler, maxReminders, backoffMs);
+        MessageService service = new MessageService(agents, injector, rendezvous, inbox, pushLoop, null, System::nanoTime);
+        return new ManualPushWiring(service, leadHerdr, scheduler);
+    }
+
     private void awaitNudge(FakeHerdr leadHerdr) throws InterruptedException {
         long deadline = System.currentTimeMillis() + 3000;
         while (!leadHerdr.called("agent.prompt") && System.currentTimeMillis() < deadline) {
@@ -1882,16 +1912,46 @@ class MessageServiceTest {
 
     @Test
     void anAlreadyCollectedTicketProducesNoNudge() throws Exception {
-        try (var wiring = wireWithPushLoop(1, 300)) { // wide backoff: poll before the first tick fires
-            String ticket = wiring.service().sendAsync(T, "long task");
-            awaitWaiting();
-            injectDelivery();
-            assertTrue(rendezvous.resolve(T, "async result"));
+        // fleetd #608: backoff is 1ms — the most hostile value there is, the tick due immediately —
+        // and the test still must pass, because with ManualScheduler the tick never runs on a timer
+        // at all; it only runs when this test calls runDueTasks() below. The old version bet a 300ms
+        // backoff was "wide" enough that collecting the ticket always won the race against a real
+        // scheduler's own timer; that held on an idle machine and failed under a loaded full-suite
+        // run — a false red on correct code, since nothing here forced that ordering, it only made
+        // it likely.
+        try (var wiring = wireWithManualScheduler(1, 1)) {
+            java.util.concurrent.CountDownLatch terminalReached = new java.util.concurrent.CountDownLatch(1);
+            // finishAsyncTask's task.future.complete(...) runs every whenComplete registered on that
+            // same future — including sendAsync's own hook that calls ReplyPushLoop.onTicketTerminal
+            // — synchronously, before complete() returns (see finishAsyncTask's javadoc and fleetd
+            // #399). This test-only hook fires right after that complete() call, on the very same
+            // (async executor) thread, so waiting for it guarantees onTicketTerminal has already run
+            // and the ticket is really sitting in the push loop's pending set — unlike waiting for
+            // Phase.DONE via poll(), which CompletableFuture.complete() can make visible to another
+            // thread before every whenComplete dependent has actually finished running (the same
+            // publish-then-run-dependents gap awaitCompletionStamped exists to close elsewhere).
+            String ticket;
+            wiring.service().setAfterFinishAsyncTaskCompleteHookForTest(terminalReached::countDown);
+            try {
+                ticket = wiring.service().sendAsync(T, "long task");
+                awaitWaiting();
+                injectDelivery();
+                assertTrue(rendezvous.resolve(T, "async result"));
+                assertTrue(terminalReached.await(5, TimeUnit.SECONDS),
+                        "the ticket never reached its terminal phase");
+            } finally {
+                wiring.service().setAfterFinishAsyncTaskCompleteHookForTest(null);
+            }
 
+            // Collect it — the exact action the nudge must never follow.
             MessageService.TaskView view = awaitTicketPhaseOn(wiring.service(), ticket, MessageService.Phase.DONE);
             assertEquals(MessageService.Phase.DONE, view.phase());
 
-            Thread.sleep(400); // let the scheduled tick run — it must find nothing pending
+            // Now run the pending tick explicitly. onTicketTerminal scheduled exactly one (the ticket
+            // is the only thing this lead has ever had pending); it must find nothing left pending —
+            // the collection above already removed it — and send no nudge.
+            assertEquals(1, wiring.scheduler().runDueTasks(),
+                    "expected exactly the one tick onTicketTerminal scheduled");
             assertFalse(wiring.leadHerdr().called("agent.prompt"),
                     "a ticket the lead already polled must never be nudged");
         }

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