Merge #367: dead lead tabs stop accumulating, without destroying a live one

fleetd #359. LeadTabScanner used to join labelled tabs straight to terminals
with no liveness check, and its javadoc excused that ("a stale name costs
nothing here"). It cost plenty: LeadCoordLoop reads that map to pick which
pane a peer message goes into, so a dead tab was a candidate it could pick.
LeadLauncher, meanwhile, had no cleanup path at all -- every reconcile that
found 0 live created another tab and left the old one.

Both now cross-check agent.list, and neither trusts a single reading of it.
That matters because the daemon's own evidence on fleet01 was agent.list
reporting 0 live while ps showed one real claude. A first cut of this fix
closed tabs on that single reading, which would have closed the operator's
live lead instead of leaving a spare tab. So: a dead tab is flagged, not
closed, and only closed when a later reconcile still finds it dead; and the
scanner grants one grace scan to a terminal it already knew was live.

Verified on this merge, not taken from the worker's report:
  mvn clean install -> Tests run: 1412, Failures: 0, Errors: 0, BUILD SUCCESS

Mutation run on merge (PendingCloseMarker.strip made identity, so a flagged
tab stops matching its configured label): 4 failures, BUILD FAILURE.

Known and accepted: ensureLeads() runs at startup, so the second reading
arrives at the next restart. A tab that dies mid-session stays flagged and
open until then. Deliberate -- the bug is about repeated restarts, and one
leftover tab is cheaper than closing a live session on unverified evidence.

fleetd/src/main/java/dev/ltms/fleet/herdr/LeadTabScanner.java

@@ -5,6 +5,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Locale;
 import java.util.Map;
@@ -53,17 +54,40 @@ import java.util.function.Supplier;
  * daemon and found again by this scan. The trust direction above is unaffected — fleetd writing a
  * name for a lead it just started is not a pane promoting itself — but <em>staleness</em> becomes
  * real: a label left behind by a session that has since died would read as a live lead forever.
- * This scanner does not solve that (its job is naming, and a stale name costs nothing here); the
- * launcher does, by requiring a running agent in the tab before it counts the lead as live. If you
- * ever make a decision that <em>removes</em> something based on this map, add the same check.
  * The remaining hazard is an <em>operator</em> one — a worker {@code tabLabel} template that
  * happens to start with the same prefix would promote the whole fleet — and that is refused at
  * startup by {@code FleetConfig.validateLeadTabPrefixes} rather than documented here.
  *
+ * <p><strong>fleetd #359 — the staleness check this class used to skip.</strong> This used to say
+ * "a stale name costs nothing here" and leave liveness to {@code LeadLauncher}, on the theory that
+ * naming and removing are different decisions. That was wrong: {@code dev.ltms.fleet.msg.LeadCoordLoop}
+ * makes exactly the kind of removal decision the old javadoc warned about, by reading this map to
+ * pick which pane a peer message goes into — and a stale entry there is not free. On a host where
+ * {@code fleetd} had restarted more than once, a labelled-but-dead tab from a previous life was
+ * reported right alongside the live one; {@code LeadCoordLoop.resolveLocalLead()} saw more than one
+ * candidate and refused to guess (safe), but the fix the daemon's own WARN suggests — name a lead
+ * after {@code coordinator.selfId} — stops being safe once two tabs can share a label: step 1 of
+ * that resolution picks whichever matching entry it finds first, which can be the dead one, and
+ * typing a peer's message into a dead shell does not fail — it is silently gone instead of merely
+ * held. {@link #scan()} now cross-checks every labelled tab against {@code agent.list} (the same
+ * signal {@code LeadLauncher.countLeads} already trusts for the same purpose) and drops any tab
+ * with no agent running in it, so a dead tab is never in the map for a caller to pick at all.
+ *
  * <p><strong>Caching.</strong> {@link #get()} is on the request path (every resolve), so the scan
- * is TTL-cached and a stale-but-valid map is preferred to a herdr round-trip. A failed scan keeps
- * the previous answer instead of emptying it — a herdr hiccup must not silently demote a live lead
- * mid-session.
+ * is TTL-cached and a stale-but-valid map is preferred to a herdr round-trip. A failed scan (herdr
+ * throws) keeps the previous answer instead of emptying it — a herdr hiccup must not silently
+ * demote a live lead mid-session.
+ *
+ * <p><strong>fleetd #359 review, finding 2 — a successful-but-wrong scan is the same hazard.</strong>
+ * The catch above only fires when a call throws. It does nothing for a call that returns 200 with an
+ * incomplete answer — exactly what the ticket's own evidence showed {@code agent.list} can do. Once
+ * this class started trusting that signal, an empty read would otherwise get cached as fact and
+ * silently drop a lead {@code CallerResolver} had, until then, correctly resolved — turning it into a
+ * {@code Role.WORKER}, which refuses every orchestration call. So a terminal this class already
+ * reported as live is not dropped the first time {@code agent.list} loses it: {@link #scan()} grants
+ * it one grace scan (see {@code gracedTerminals}) and only drops it if a <em>later</em> scan still
+ * finds no agent. A terminal never reported live before gets no grace — that would weaken the
+ * original #359 fix itself, which this class's own test suite already pins.
  */
 public final class LeadTabScanner implements Supplier<Map<String, String>> {
 
@@ -79,6 +103,15 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
     private long scannedAtNanos;
     private boolean everScanned;
 
+    /**
+     * Terminals currently on their one grace scan: {@code cached} reported them live, the most
+     * recent {@link #scan()} found no agent for them, and they were re-included anyway. Cleared for
+     * a terminal the instant it is seen live again; a terminal still here on the <em>next</em> scan
+     * is finally dropped. Scoped separately from {@link #cached} so a graced terminal cannot renew
+     * its own grace forever just by staying in the exposed map (fleetd #359 review, finding 2).
+     */
+    private Set<String> gracedTerminals = Set.of();
+
     /**
      * @param herdr                   the herdr client to query ({@code workspace.list},
      *                                {@code tab.list}, {@code pane.list} — all read-only)
@@ -141,7 +174,7 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
         return cached;
     }
 
-    /** One full pass: labelled tabs → their panes → those panes' terminals. */
+    /** One full pass: labelled tabs → live agents in them → those panes' terminals. */
     private Map<String, String> scan() {
         Map<String, String> nameByTab = new LinkedHashMap<>();
         for (JsonNode w : herdr.call("workspace.list").path("workspaces")) {
@@ -158,17 +191,47 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
             }
         }
 
-        Map<String, String> byTerminal = new LinkedHashMap<>();
-        if (!nameByTab.isEmpty()) {
-            // One pane.list for every tab: panes carry tab_id, so the join is local.
-            for (JsonNode p : herdr.call("pane.list", Map.of()).path("panes")) {
-                String name = nameByTab.get(p.path("tab_id").asText(null));
-                String terminal = p.path("terminal_id").asText(null);
-                if (name != null && terminal != null && !terminal.isBlank()) {
-                    byTerminal.put(terminal, name);
-                }
+        if (nameByTab.isEmpty()) {
+            gracedTerminals = Set.of();
+            return Map.of();
+        }
+
+        // fleetd #359: a labelled tab is only a lead when herdr also reports a running agent in
+        // it — the same liveness signal LeadLauncher.countLeads trusts for the identical purpose.
+        // Without this, a tab left behind by a session that has since died reads as live forever.
+        Set<String> tabsWithAgent = new HashSet<>();
+        for (JsonNode a : herdr.call("agent.list").path("agents")) {
+            String tabId = a.path("tab_id").asText(null);
+            if (tabId != null) {
+                tabsWithAgent.add(tabId);
             }
         }
+
+        Map<String, String> byTerminal = new LinkedHashMap<>();
+        Set<String> stillGraced = new HashSet<>();
+        // One pane.list for every tab: panes carry tab_id, so the join is local.
+        for (JsonNode p : herdr.call("pane.list", Map.of()).path("panes")) {
+            String tabId = p.path("tab_id").asText(null);
+            String name = nameByTab.get(tabId);
+            String terminal = p.path("terminal_id").asText(null);
+            if (name == null || terminal == null || terminal.isBlank()) {
+                continue;
+            }
+            if (tabsWithAgent.contains(tabId)) {
+                byTerminal.put(terminal, name);
+                continue;
+            }
+            // No agent reported for this tab, but its tab/pane are still here — this is the
+            // ambiguous case review finding 2 named: a successful agent.list that came back short
+            // does not prove the lead is dead. Grant one grace scan to a terminal we had already
+            // reported as live; a terminal we never reported live gets none, so the original #359
+            // fix (a genuinely dead tab is never reported) is unaffected for the common case.
+            if (cached.containsKey(terminal) && !gracedTerminals.contains(terminal)) {
+                byTerminal.put(terminal, name);
+                stillGraced.add(terminal);
+            }
+        }
+        gracedTerminals = stillGraced;
         return Collections.unmodifiableMap(byTerminal);
     }
 
@@ -177,12 +240,14 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
      *
      * <p>Exact match (case-insensitive, ends stripped) against {@link #tabToName} — no prefix
      * stripping, so an operator's {@code "lead: something-else"} tab is never mistaken for a
-     * configured lead just because it shares a prefix.
+     * configured lead just because it shares a prefix. The match strips a trailing
+     * {@link PendingCloseMarker} first, so a tab {@code LeadLauncher} has flagged as maybe-dead but
+     * not yet closed keeps resolving normally while that reconcile is pending.
      */
     private String leadNameOf(String label) {
         if (label == null) {
             return null;
         }
-        return tabToName.get(label.strip().toLowerCase(Locale.ROOT));
+        return tabToName.get(PendingCloseMarker.strip(label).toLowerCase(Locale.ROOT));
     }
 }

fleetd/src/main/java/dev/ltms/fleet/herdr/PendingCloseMarker.java

@@ -0,0 +1,42 @@
+package dev.ltms.fleet.herdr;
+
+/**
+ * The suffix {@code dev.ltms.fleet.lead.LeadLauncher} appends to a lead tab's label the first time a
+ * reconcile finds no running agent in it, before it is sure enough to close the tab outright.
+ *
+ * <p><strong>fleetd #359 review, finding 1.</strong> The daemon's own evidence showed
+ * {@code agent.list} can read "no agent" for a tab that genuinely has one running — so a single such
+ * reading must never be treated as proof a tab is dead. {@code LeadLauncher} now writes this marker
+ * on the first miss, and only closes the tab if a <em>later</em>, independent reconcile still finds
+ * it dead while the marker is still there. Two consecutive misses, one restart apart, is a much
+ * stronger claim than one.
+ *
+ * <p>{@link LeadTabScanner} strips the same suffix before matching a label against a configured
+ * lead's {@code tab}, so a flagged-but-actually-still-live tab keeps resolving normally — the marker
+ * changes nothing about which pane {@code LeadCoordLoop} can reach while the flag is pending. Both
+ * classes must use exactly this suffix, which is why it lives here rather than as a private constant
+ * on either.
+ */
+public final class PendingCloseMarker {
+
+    public static final String SUFFIX = " [fleetd:pending-close]";
+
+    private PendingCloseMarker() {
+    }
+
+    /** The label with any trailing pending-close marker removed, for name matching. */
+    public static String strip(String label) {
+        if (label == null) {
+            return null;
+        }
+        String stripped = label.strip();
+        return stripped.endsWith(SUFFIX)
+                ? stripped.substring(0, stripped.length() - SUFFIX.length()).strip()
+                : stripped;
+    }
+
+    /** Whether a label currently carries the marker. */
+    public static boolean isFlagged(String label) {
+        return label != null && label.strip().endsWith(SUFFIX);
+    }
+}

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

@@ -4,6 +4,7 @@ import dev.ltms.fleet.config.FleetConfig;
 import dev.ltms.fleet.herdr.Agent;
 import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.HerdrException;
+import dev.ltms.fleet.herdr.PendingCloseMarker;
 import dev.ltms.fleet.herdr.Tab;
 import dev.ltms.fleet.herdr.Workspace;
 import dev.ltms.fleet.herdr.WorkspaceControl;
@@ -12,9 +13,11 @@ import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import dev.ltms.fleet.peer.PeerLauncher;
 
 /**
@@ -44,8 +47,33 @@ import dev.ltms.fleet.peer.PeerLauncher;
  * privilege escalation (the tab label never granted anything a pane could take for itself; see that
  * class's javadoc), but <em>staleness</em>: a label left behind by a crashed session would otherwise
  * read as a live lead forever, and the lead would never be relaunched. So a lead counts as live only
- * when herdr also reports a <em>running agent</em> in that tab — see {@link #liveLeads}. A labelled
+ * when herdr also reports a <em>running agent</em> in that tab — see {@link #countLeads}. A labelled
  * tab with no agent in it is not a lead.
+ *
+ * <p><strong>fleetd #359 — a stale label used to pile up, not just mislead.</strong> Finding "not
+ * live" here used to mean only one thing: launch another. The old label was left exactly where it
+ * was, so a daemon that restarted enough times — or hit one herdr read that missed a genuinely
+ * running agent — accumulated one more identically-labelled dead tab per occurrence, and
+ * {@code LeadTabScanner} (before its own #359 fix) reported every one of them as a lead.
+ *
+ * <p><strong>Review finding 1 — closing on one reading is worse than the bug.</strong> The first
+ * version of this fix closed a name's dead tabs the moment a single {@link #countLeads} reading
+ * called them dead. The ticket's own live evidence rules that out: on a real host, {@code
+ * agent.list} was seen reporting "0 live" for a tab that a plain {@code ps} confirmed was running a
+ * real session. Closing on that reading would have destroyed the operator's actual lead — a worse
+ * failure than the extra tab it replaces. So {@link #ensureLeads()} now needs the same dead reading
+ * <em>twice</em>, one restart apart, before it closes anything: the first time a labelled tab reads
+ * dead, it is only flagged ({@link PendingCloseMarker}), left running, untouched; it is closed only
+ * if a <em>later</em>, independently-connected reconcile still finds it dead while the flag is
+ * still there. A transient miss self-heals — the next reconcile sees the agent again and clears the
+ * flag (see {@code toUnflag} below) — so the worst case for a single bad reading is one extra tab
+ * surviving one more restart, never a live session destroyed. Two alternatives were considered and
+ * rejected: corroborating {@code agent.list} against a second, truly independent signal was dropped
+ * because nothing else herdr exposes proves "is a process attached to this pane" any better — a
+ * second call to the same unreliable source is not independent evidence; capping the close to "all
+ * but the most recent dead tab" was dropped because "most recent" has no reliable ordering across
+ * tab ids and would leave the true failure mode (a name that is <em>never</em> reconfirmed) growing
+ * by one tab per bad reading forever, which is the exact defect this ticket exists to fix.
  */
 public final class LeadLauncher {
 
@@ -79,9 +107,9 @@ public final class LeadLauncher {
             return 0;
         }
 
-        Map<String, Integer> live;
+        Map<String, LeadCount> live;
         try {
-            live = liveLeads(leaders);
+            live = countLeads(leaders);
         } catch (HerdrException e) {
             // Counting is the whole safety mechanism against double-spawning. If we cannot count, we
             // must not guess — spawning a second orchestrator is worse than starting none.
@@ -93,9 +121,53 @@ public final class LeadLauncher {
         for (Map.Entry<String, FleetConfig.Leader> e : leaders.entrySet()) {
             String name = e.getKey();
             FleetConfig.Leader lead = e.getValue();
-            int running = live.getOrDefault(name, 0);
+            LeadCount state = live.getOrDefault(name, LeadCount.NONE);
+            int running = state.running();
             int wanted = lead.instances();
 
+            // fleetd #359 review finding 1: a tab already flagged pending-close, still labelled for
+            // this lead, and STILL hosting no agent on this separate reconcile — two independent
+            // readings agree, so close it. A tab found dead for the first time is only flagged below,
+            // never closed on the spot.
+            for (String tabId : state.toClose()) {
+                log.info("lead '{}': closing tab {} — flagged pending-close on a previous reconcile "
+                        + "and still no agent running in it", name, tabId);
+                try {
+                    spaces.closeTab(tabId);
+                } catch (RuntimeException cleanup) {
+                    log.warn("could not close stale tab {} for lead '{}': {}",
+                            tabId, name, cleanup.getMessage());
+                }
+            }
+            // A tab labelled for this lead, with no agent running in it, seen dead for the first
+            // time — flag it rather than closing it. One reading of `agent.list` is not enough
+            // evidence to destroy a tab that might genuinely be live (see the class javadoc).
+            for (String tabId : state.toFlag()) {
+                String flagged = lead.tabLabel() + PendingCloseMarker.SUFFIX;
+                log.info("lead '{}': tab {} has no agent running in it this reconcile — flagging it "
+                        + "'{}' rather than closing; it is only closed if a later reconcile still "
+                        + "finds it dead", name, tabId, flagged);
+                try {
+                    spaces.renameTab(tabId, flagged);
+                } catch (RuntimeException cleanup) {
+                    log.warn("could not flag stale tab {} for lead '{}': {}",
+                            tabId, name, cleanup.getMessage());
+                }
+            }
+            // A previously-flagged tab that is running an agent again — the miss that flagged it was
+            // transient. Clear the flag so a future, unrelated miss starts its own two-reading count
+            // rather than closing on the strength of this one's already-spent flag.
+            for (String tabId : state.toUnflag()) {
+                log.info("lead '{}': tab {} is running an agent again — clearing its pending-close flag",
+                        name, tabId);
+                try {
+                    spaces.renameTab(tabId, lead.tabLabel());
+                } catch (RuntimeException cleanup) {
+                    log.warn("could not clear the pending-close flag on tab {} for lead '{}': {}",
+                            tabId, name, cleanup.getMessage());
+                }
+            }
+
             if (running >= wanted) {
                 log.info("lead '{}': {} live, {} wanted — nothing to start", name, running, wanted);
                 continue;
@@ -126,18 +198,29 @@ public final class LeadLauncher {
     }
 
     /**
-     * How many live leads exist per configured name: a running agent in a tab labelled with that
-     * lead's exact {@code tab} (CB-579). Member workspaces are excluded, exactly as the scanner
-     * excludes them: a member must not be counted as a lead because it happens to sit in a matching
-     * tab.
+     * How many live leads exist per configured name, and which of that name's labelled tabs are
+     * <em>not</em> live: a running agent in a tab labelled with that lead's exact {@code tab}
+     * (CB-579). Member workspaces are excluded, exactly as the scanner excludes them: a member must
+     * not be counted as a lead because it happens to sit in a matching tab.
      *
      * <p>There used to be a second path here — a running agent on the terminal a
      * {@code fleet.leaders.<name>.terminal} pin named, for a lead opened and pinned by hand. That
      * pin is retired: {@code tab} is now the only field identity depends on, and {@link Agent}
      * already carries {@link Agent#tabId()} directly, so a hand-opened lead is found the same way an
      * auto-launched one is — by labelling its tab to match.
+     *
+     * <p>fleetd #359 review finding 1: a labelled tab with nothing running in it is split into
+     * {@code toClose} (already flagged pending-close by a previous reconcile, and still dead — two
+     * independent readings agree) and {@code toFlag} (dead for the first time — not enough evidence
+     * to close yet). {@code toUnflag} is the reverse: a tab flagged pending-close that is running an
+     * agent again, so the flag it carries no longer means anything and {@link #ensureLeads()} clears
+     * it.
      */
-    private Map<String, Integer> liveLeads(Map<String, FleetConfig.Leader> leaders) {
+    private record LeadCount(int running, List<String> toClose, List<String> toFlag, List<String> toUnflag) {
+        static final LeadCount NONE = new LeadCount(0, List.of(), List.of(), List.of());
+    }
+
+    private Map<String, LeadCount> countLeads(Map<String, FleetConfig.Leader> leaders) {
         // A lead and the members share ONE workspace now (the operator asked for a single "session"
         // with many tabs), so a workspace can no longer be excluded wholesale — the lead lives in the
         // member workspace by design. The sole discriminator is the exact tab label: a lead carries
@@ -145,6 +228,7 @@ public final class LeadLauncher {
         // profile's `worker: {profile} #{n}` template. These never collide, so an exact-label match
         // separates them without needing to know which workspace anyone is in.
         Map<String, String> nameByTab = new LinkedHashMap<>();
+        Set<String> flaggedTabIds = new LinkedHashSet<>();
         for (Workspace ws : spaces.listWorkspaces()) {
             if (ws.workspaceId() == null) {
                 continue;
@@ -153,31 +237,65 @@ public final class LeadLauncher {
                 String declared = leadNameOf(tab.label(), leaders);
                 if (declared != null && tab.tabId() != null) {
                     nameByTab.put(tab.tabId(), declared);
+                    if (PendingCloseMarker.isFlagged(tab.label())) {
+                        flaggedTabIds.add(tab.tabId());
+                    }
                 }
             }
         }
 
+        Set<String> liveTabIds = new LinkedHashSet<>();
         Map<String, Integer> counts = new LinkedHashMap<>();
         for (Agent a : agents.list()) {
             String name = nameByTab.get(a.tabId());
             if (name != null) {
                 counts.merge(name, 1, Integer::sum);
+                liveTabIds.add(a.tabId());
             }
         }
-        return counts;
+
+        Map<String, List<String>> toCloseByName = new LinkedHashMap<>();
+        Map<String, List<String>> toFlagByName = new LinkedHashMap<>();
+        Map<String, List<String>> toUnflagByName = new LinkedHashMap<>();
+        nameByTab.forEach((tabId, name) -> {
+            boolean live = liveTabIds.contains(tabId);
+            boolean flagged = flaggedTabIds.contains(tabId);
+            if (live) {
+                if (flagged) {
+                    toUnflagByName.computeIfAbsent(name, k -> new ArrayList<>()).add(tabId);
+                }
+                return;
+            }
+            if (flagged) {
+                toCloseByName.computeIfAbsent(name, k -> new ArrayList<>()).add(tabId);
+            } else {
+                toFlagByName.computeIfAbsent(name, k -> new ArrayList<>()).add(tabId);
+            }
+        });
+
+        Map<String, LeadCount> out = new LinkedHashMap<>();
+        for (String name : leaders.keySet()) {
+            out.put(name, new LeadCount(counts.getOrDefault(name, 0),
+                    toCloseByName.getOrDefault(name, List.of()),
+                    toFlagByName.getOrDefault(name, List.of()),
+                    toUnflagByName.getOrDefault(name, List.of())));
+        }
+        return out;
     }
 
     /**
      * The configured lead a tab label names, or {@code null} for a label that names none.
      *
      * <p>Matched exactly (case-insensitively) against each lead's configured {@code tab}, so an
-     * operator's {@code "lead: something-else"} tab is not mistaken for a configured lead.
+     * operator's {@code "lead: something-else"} tab is not mistaken for a configured lead. A
+     * trailing {@link PendingCloseMarker} is stripped first, so a tab this class flagged on a
+     * previous reconcile is still recognised as the same lead's tab on this one.
      */
     private String leadNameOf(String label, Map<String, FleetConfig.Leader> leaders) {
         if (label == null) {
             return null;
         }
-        String l = label.strip();
+        String l = PendingCloseMarker.strip(label);
         for (Map.Entry<String, FleetConfig.Leader> e : leaders.entrySet()) {
             String tab = e.getValue().tabLabel();
             if (tab != null && l.equalsIgnoreCase(tab.strip())) {

fleetd/src/test/java/dev/ltms/fleet/herdr/LeadTabScannerTest.java

@@ -6,6 +6,7 @@ import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -35,6 +36,12 @@ class LeadTabScannerTest {
         final Map<String, String[]> tabs = new LinkedHashMap<>();
         /** pane_id → [tab_id, terminal_id]. */
         final Map<String, String[]> panes = new LinkedHashMap<>();
+        /**
+         * tab_id → whether herdr reports a running agent there. Defaults to {@code true} for every
+         * tab that has a pane, so every existing fixture keeps meaning "a live lead" unless a test
+         * says otherwise via {@link #deadAgent}.
+         */
+        final Set<String> deadTabs = new LinkedHashSet<>();
         int calls;
         boolean failing;
 
@@ -53,6 +60,18 @@ class LeadTabScannerTest {
             return this;
         }
 
+        /** Mark {@code tabId} as labelled but agent-less — a dead lead's leftover tab (fleetd #359). */
+        TopologyHerdr deadAgent(String tabId) {
+            deadTabs.add(tabId);
+            return this;
+        }
+
+        /** Undo {@link #deadAgent} — models {@code agent.list} reporting the tab live again. */
+        TopologyHerdr reviveAgent(String tabId) {
+            deadTabs.remove(tabId);
+            return this;
+        }
+
         @Override
         public JsonNode call(String method, Object params) {
             calls++;
@@ -83,6 +102,19 @@ class LeadTabScannerTest {
                                     .formatted(id, p[0], p[1])));
                     return read("{\"panes\":[%s]}".formatted(String.join(",", items)));
                 }
+                case "agent.list" -> {
+                    // One agent per distinct tab that has a pane and isn't marked dead — mirrors
+                    // AgentControl.list()'s "agents" shape closely enough for the scanner's join,
+                    // which only reads tab_id off each entry.
+                    Set<String> seen = new LinkedHashSet<>();
+                    panes.forEach((paneId, p) -> {
+                        String tabId = p[0];
+                        if (!deadTabs.contains(tabId) && seen.add(tabId)) {
+                            items.add("{\"tab_id\":\"%s\"}".formatted(tabId));
+                        }
+                    });
+                    return read("{\"agents\":[%s]}".formatted(String.join(",", items)));
+                }
                 default -> throw new AssertionError("unexpected herdr call: " + method);
             }
         }
@@ -208,6 +240,114 @@ class LeadTabScannerTest {
                 scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get().get("term_opus_split"));
     }
 
+    // ── fleetd #359: a labelled tab is only a lead when something is running in it ─────────────────
+
+    /**
+     * The core invariant this ticket restores: {@code get()} must never report a terminal for a
+     * lead whose pane no longer runs an agent. Before this fix the scanner joined labelled tabs to
+     * panes with no liveness check at all, so a tab left behind by a crashed/relaunched lead (see
+     * {@code LeadLauncher}'s own staleness handling) was reported as live forever — which is exactly
+     * what let duplicate lead tabs make {@code LeadCoordLoop.resolveLocalLead()} permanently unable
+     * to pick one. Mutate this away (drop the {@code agent.list} cross-check in {@link
+     * LeadTabScanner#scan()}) and this test must fail.
+     */
+    @Test
+    void aLabelledTabWithNoRunningAgentIsNotReported() {
+        TopologyHerdr herdr = twoLeads().deadAgent("w1:t1"); // opus-5.0's tab is labelled but dead
+
+        Map<String, String> leads = scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get();
+
+        assertFalse(leads.containsKey("term_opus"),
+                "a labelled tab with no running agent must never be reported as a live lead");
+        assertEquals("gpt-sol-5.6", leads.get("term_gpt"),
+                "the other, genuinely live lead must be unaffected");
+    }
+
+    /**
+     * The other direction, pinned separately so a fix cannot satisfy the test above by simply
+     * returning nothing: a labelled tab that DOES have a running agent must still be reported. A
+     * scanner that always comes back empty is worse than the bug it fixes.
+     */
+    @Test
+    void aLabelledTabWithARunningAgentIsStillReported() {
+        Map<String, String> leads = scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong()).get();
+
+        assertEquals("opus-5.0", leads.get("term_opus"));
+        assertEquals("gpt-sol-5.6", leads.get("term_gpt"));
+    }
+
+    /**
+     * fleetd #359 review, finding 2 — the exact scenario the ticket's own evidence showed:
+     * {@code agent.list} can come back successfully but short, without the herdr call ever throwing.
+     * A lead this class already reported as live must not be dropped on the strength of one such
+     * read: {@link LeadTabScanner#get()}'s "keep the cache on failure" contract only fires on an
+     * exception, so without a fix a single short {@code agent.list} silently empties the cached map —
+     * which would resolve that lead's pane as {@code Role.WORKER} downstream, refusing every
+     * orchestration call. Mutate this away (drop the one-scan grace in {@link
+     * LeadTabScanner#scan()}) and this test must fail.
+     */
+    @Test
+    void aTransientAgentListMissDoesNotDemoteALeadAlreadyKnownLive() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
+        assertTrue(s.get().containsKey("term_opus"), "opus must be known live before the miss");
+
+        // One scan where agent.list comes back without opus's tab, even though the tab and pane are
+        // completely unchanged — the tab/pane are still there, only the liveness read is short.
+        herdr.deadAgent("w1:t1");
+        clock.addAndGet(TTL);
+
+        assertTrue(s.get().containsKey("term_opus"),
+                "a single missed detection must not empty the cached map for a lead already known live");
+        assertEquals("gpt-sol-5.6", s.get().get("term_gpt"), "the unaffected lead is unchanged");
+    }
+
+    /**
+     * The other half of finding 2, so the grace above cannot be mistaken for permanent amnesty: the
+     * original #359 invariant (a genuinely dead tab is not reported forever) must still hold once a
+     * SECOND, independent scan agrees the agent is gone.
+     */
+    @Test
+    void aLeadMissingFromAgentListOnTwoConsecutiveScansIsFinallyDropped() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
+        assertTrue(s.get().containsKey("term_opus"));
+
+        herdr.deadAgent("w1:t1");
+        clock.addAndGet(TTL);
+        assertTrue(s.get().containsKey("term_opus"), "first miss is a grace period, not a verdict");
+
+        clock.addAndGet(TTL); // a second, independent scan — still no agent
+        assertFalse(s.get().containsKey("term_opus"),
+                "a second consecutive miss for the same terminal must finally drop it");
+    }
+
+    /** A lead that recovers between the two misses keeps its grace spent, not renewed for free. */
+    @Test
+    void aLeadThatRecoversBetweenMissesIsReportedNormallyAndResetsItsGrace() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
+        assertTrue(s.get().containsKey("term_opus"));
+
+        herdr.deadAgent("w1:t1");
+        clock.addAndGet(TTL);
+        assertTrue(s.get().containsKey("term_opus"), "graced on the first miss");
+
+        herdr.reviveAgent("w1:t1"); // the miss really was transient
+        clock.addAndGet(TTL);
+        assertTrue(s.get().containsKey("term_opus"), "found live again — reported normally");
+
+        // A later, unrelated miss must get its own fresh grace scan rather than being dropped
+        // immediately because the earlier miss had already "used up" a slot for this terminal.
+        herdr.deadAgent("w1:t1");
+        clock.addAndGet(TTL);
+        assertTrue(s.get().containsKey("term_opus"),
+                "a miss after a genuine recovery is a new event and deserves its own grace scan");
+    }
+
     /**
      * CB-579 acceptance (6): this is the bug the ticket closes. A stale pin used to be merged back
      * over every scan and never expire; now a scan is the whole answer, so a lead whose tab is gone

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

@@ -9,6 +9,7 @@ import org.junit.jupiter.api.Test;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -106,6 +107,7 @@ class LeadLauncherTest {
 
         assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
         assertFalse(herdr.called("agent.start"), "the live lead must not be duplicated");
+        assertFalse(herdr.called("tab.close"), "a labelled tab WITH a live agent must never be closed");
     }
 
     /**
@@ -122,6 +124,121 @@ class LeadLauncherTest {
                 "a stale label is not a lead; the lead must be relaunched");
     }
 
+    /**
+     * fleetd #359 review finding 1 — the exact scenario the ticket's own live evidence produced: on a
+     * real host, {@code agent.list} reported "0 live" for a tab a plain {@code ps} confirmed was
+     * running a real session. A single such reading must never close the tab outright — that would
+     * destroy the operator's actual lead, a worse failure than the stale-tab bug this ticket exists
+     * to fix. The first dead reading only flags the tab; mutate this away (make the first reading
+     * close instead of flag) and this test must fail.
+     */
+    @Test
+    void aStaleLabelledTabIsFlaggedRatherThanClosedOnTheFirstReconcile() {
+        FakeHerdr herdr = new FakeHerdr()
+                .withWorkspace("wL", "fleet")
+                .withTab("wL", "wL:t1", "lead: opus"); // label only, first look — could be a live session agent.list missed
+
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
+
+        assertFalse(herdr.called("tab.close"),
+                "one missed reading must never close a tab that might be hosting a live session");
+        assertTrue(flaggedTabIds(herdr).contains("wL:t1"),
+                "the stale tab must be flagged pending-close so a later reconcile can confirm it");
+    }
+
+    /**
+     * The operator's own trace on fleet01 (#359): a daemon that restarted several times, each
+     * occasion finding "0 live" for whatever reason, had left several identically-labelled dead
+     * tabs sitting side by side. Every one of them is flagged on its first dead reading, not closed —
+     * none is more or less trustworthy than another.
+     */
+    @Test
+    void allStaleLabelledTabsAreFlaggedRatherThanClosedOnTheFirstReconcile() {
+        FakeHerdr herdr = new FakeHerdr()
+                .withWorkspace("wL", "fleet")
+                .withTab("wL", "wL:t1", "lead: opus")
+                .withTab("wL", "wL:t2", "lead: opus")
+                .withTab("wL", "wL:t3", "lead: opus"); // three restarts' worth of debris, none confirmed twice yet
+
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
+
+        assertFalse(herdr.called("tab.close"), "no tab may be closed on its first dead reading");
+        assertEquals(Set.of("wL:t1", "wL:t2", "wL:t3"), flaggedTabIds(herdr),
+                "every dead labelled tab must be flagged, not just the first one found");
+    }
+
+    /**
+     * fleetd #359 review finding 1, the other half — a tab already flagged pending-close by an
+     * earlier reconcile, and STILL dead on this one, has now been read dead on two independent,
+     * separately-connected reconciles. That is strong enough evidence to actually close it. Mutate
+     * this away (never close a flagged tab) and this test must fail — the original #359 growth bug
+     * would come back for good.
+     */
+    @Test
+    void aTabAlreadyFlaggedPendingCloseIsClosedWhenStillDeadOnALaterReconcile() {
+        FakeHerdr herdr = new FakeHerdr()
+                .withWorkspace("wL", "fleet")
+                .withTab("wL", "wL:t1", "lead: opus [fleetd:pending-close]"); // flagged last reconcile, still dead
+
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
+
+        assertTrue(herdr.called("tab.close"),
+                "a tab dead on two independent reconciles must finally be closed");
+        assertEquals("wL:t1", ((Map<?, ?>) herdr.lastCall("tab.close").params()).get("tab_id"));
+    }
+
+    /** All of several already-flagged, still-dead tabs are closed — not just the first found. */
+    @Test
+    void allTabsAlreadyFlaggedPendingCloseAreClosedWhenStillDead() {
+        FakeHerdr herdr = new FakeHerdr()
+                .withWorkspace("wL", "fleet")
+                .withTab("wL", "wL:t1", "lead: opus [fleetd:pending-close]")
+                .withTab("wL", "wL:t2", "lead: opus [fleetd:pending-close]")
+                .withTab("wL", "wL:t3", "lead: opus [fleetd:pending-close]");
+
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
+
+        List<Object> closedTabIds = herdr.calls.stream()
+                .filter(c -> c.method().equals("tab.close"))
+                .<Object>map(c -> ((Map<?, ?>) c.params()).get("tab_id"))
+                .toList();
+        assertEquals(3, closedTabIds.size(),
+                "every confirmed-dead labelled tab must be closed, not just the first one found");
+        assertEquals(Set.of("wL:t1", "wL:t2", "wL:t3"), Set.copyOf(closedTabIds));
+    }
+
+    /**
+     * A tab flagged pending-close on a previous reconcile that is running an agent again — the miss
+     * that flagged it was transient. It must never be closed, and its flag must be cleared so a
+     * future, unrelated miss starts its own two-reading count from zero.
+     */
+    @Test
+    void aFlaggedTabRunningAnAgentAgainHasItsFlagClearedInsteadOfBeingClosed() {
+        FakeHerdr herdr = new FakeHerdr()
+                .withWorkspace("wL", "fleet")
+                .withTab("wL", "wL:t1", "lead: opus [fleetd:pending-close]")
+                .withAgent("lead-opus", "term_lead", "wL:p1", "wL:t1"); // it recovered — really alive now
+
+        assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
+                "the lead is live again — nothing to relaunch");
+
+        assertFalse(herdr.called("tab.close"), "a tab running an agent again must never be closed");
+        assertFalse(herdr.called("agent.start"), "the lead is live again — nothing to relaunch");
+        assertEquals("wL:t1", ((Map<?, ?>) herdr.lastCall("tab.rename").params()).get("tab_id"));
+        assertEquals("lead: opus", ((Map<?, ?>) herdr.lastCall("tab.rename").params()).get("label"),
+                "the pending-close flag must be cleared once the tab is confirmed live again");
+    }
+
+    /** Every {@code tab.rename} call whose label carries the pending-close marker, by tab id. */
+    private static Set<String> flaggedTabIds(FakeHerdr herdr) {
+        return herdr.calls.stream()
+                .filter(c -> c.method().equals("tab.rename"))
+                .filter(c -> String.valueOf(((Map<?, ?>) c.params()).get("label"))
+                        .endsWith("[fleetd:pending-close]"))
+                .map(c -> String.valueOf(((Map<?, ?>) c.params()).get("tab_id")))
+                .collect(java.util.stream.Collectors.toUnmodifiableSet());
+    }
+
     /**
      * A lead the operator opened by hand is live once its tab carries the configured `tab:` label —
      * CB-579 retired the `terminal:` pin, so a hand-opened lead is found the same way an