CB-578 stage A: classify a usage-limit refusal instead of a completed reply

A backend that refuses on a subscription usage limit leaves the pane healthy but the turn ends with no bridge_reply; the completion fallback used to scrape and hand that refusal back as if it were a real answer. CompletionResolver now matches the scrape against a per-profile exhaustedPattern (config, never a vendor string) and resolves the send as Rendezvous.Kind/Outcome.BACKEND_EXHAUSTED with a reason carrying the matched line, kept distinct from GONE/WORKER_FAILED. A profile with no pattern configured is unaffected. Coverage is logged at startup via CompletionResolver.coverage(...), naming which profiles have a pattern and which don't, following FleetHealthMonitor.coverage's pattern.
Merge CB-579: resolve a lead by its tab name, drop the terminal-id pin
2026-08-15 09:55:06 +02:00 · 2026-08-15 09:29:00 +02:00 · 2026-08-15 08:57:03 +02:00 · 2026-08-15 08:54:33 +02:00 · 2026-08-15 08:47:12 +02:00 · 2026-08-15 08:05:21 +02:00
34 changed files with 1108 additions and 365 deletions
@@ -47,15 +47,18 @@ bind:
 # (say a Claude lead and an opencode lead) work as peers: the second is silently demoted and refused
 # every orchestration call. List each lead's pane here and all of them resolve as leads.
 #
-#   terminal → the ONLY field identity depends on; get it from that session's bridge_whoami
+#   tab → the ONLY field identity depends on (CB-579); the exact label of the tab hosting the lead.
+#         Label the tab yourself, or let bridged label one it launches — see `fleet.leaders:` below.
 #   kind/model → descriptive; they document what runs in the pane and are echoed by bridge_whoami
 #
-# A lead is never spawned — it pre-exists, which is exactly why it must be named rather than created.
+# A lead's tab must already carry its label (or be launched by bridged, which labels it) — there is
+# no terminal id to paste in and nothing to re-pin when the session restarts: the tab survives, so
+# the same label resolves the same lead again on the next scan.
 # `bridge_whoami` reports `{"role":"primary","leader":"<name>"}`; role stays "primary" because a lead
 # IS a primary for authorization, so nothing that keys on the role breaks.
 #
 # KEEP `primary:` when adding leads: it still addresses the CB-307 push loop, which needs a single
-# destination for its nudges. If both name the same terminal, the `fleet.leaders:` entry wins.
+# destination for its nudges, and is a separate mechanism from lead identity — see `fleet.leaders:`.
 #
 # Leads are configured under `fleet.leaders:` — see THE FLEET further down.
 #
@@ -135,6 +138,12 @@ herdrSocket: ~/.config/herdr/herdr.sock
 #                      SSH is unaffected). The token value itself is never stored in this file.
 #   gitHostEnv       → host env var holding the forge host (default GITEA_HOST). Injected as
 #                      GITEA_HOST *only* alongside a resolved gitTokenEnv.
+#   exhaustedPattern → regex matched against a completion-fallback scrape (CB-578 stage A) to
+#                      classify a turn that ended with no bridge_reply as the backend having
+#                      refused on a subscription usage limit, rather than a real answer. Opt-in —
+#                      omit and this profile's completion fallback behaves exactly as before.
+#                      Every backend words its refusal differently, so this is config, never a
+#                      vendor string baked into bridged itself.
 #   env              → extra environment for this profile's workers, as a literal key/value map
 #                      (CB-511). Use it to give workers a toolchain.
 #
@@ -168,6 +177,7 @@ profiles:
    maxLoad: 2                        # max live workers on this profile (omit for unlimited)
    # gitTokenEnv: GITEA_TOKEN        # opt-in: let this profile's workers open their own PR (CB-302)
    # gitHostEnv: GITEA_HOST          # defaults to GITEA_HOST; injected only with gitTokenEnv
+    # exhaustedPattern: "usage limit has been reached"   # opt-in: classify a usage-limit refusal (CB-578)
    # configDir: /Users/me/.ccs/instances/gx10   # CLAUDE_CONFIG_DIR — inherit that profile's skills/MCP
    # cwd: /Users/me/src/myrepo       # pin the working dir; omit to inherit the primary's
    # parityOverlay: [".claude/settings.local.json", ".env", ".envrc"]   # never add .mcp.json — see above
@@ -309,15 +319,18 @@ fleet:

  # Panes that orchestrate rather than are orchestrated. A lead may now be CREATED as well as
  # recognised: give it a `profile:` and the daemon launches the shortfall when fewer than
-  # `instances` are live. Give it only a `terminal:` and it is recognise-only, as before.
+  # `instances` are live. Omit `profile:` and it is recognise-only, as before.
  #
-  # `tabPrefix` is the naming convention that finds a lead without pasting a terminal id: label the
-  # tab `lead: <name>` when you open it and the pane is recognised on the next rescan. Reopen the
-  # tab later and the id changes; the label does not.
+  # `tab:` (CB-579) is REQUIRED and is the only field identity depends on — the exact label of the
+  # tab hosting the lead, matched case-insensitively. Label the tab yourself and put that same
+  # string here, and the pane is recognised on the next rescan. Reopen the tab later, or the session
+  # inside it restarts — the terminal id changes; the tab, and its label, do not, so no config edit
+  # follows a restart.
  #
-  # A lead the daemon launches is labelled BY the daemon, using the same convention, so it is found
+  # A lead the daemon launches is labelled BY the daemon with this same `tab:` value, so it is found
  # by the same scan. A lead counts as live only when herdr also reports a running agent in that
-  # tab — a label left behind by a session that died does not block the relaunch.
+  # tab — a label left behind by a session that died does not block the relaunch, and a tab that is
+  # gone entirely drops out of the next scan rather than being remembered forever.
  #
  # An auto-launched lead is NOT a member: it gets no worker reply charter, is never registered with
  # the session lifecycle (the idle reaper would kill your orchestrator), and stays on the
@@ -326,10 +339,9 @@ fleet:
  #   opus-5.0:
  #     profile: opus                # omit to never create this lead, only recognise it
  #     instances: 1                 # desired live count; only the shortfall is launched. 0 = off
-  #     terminal: term_0123456789abcd  # optional hand-pin; usually found by tabPrefix instead.
-  #                                  # A running agent on this terminal also counts as live, so a
-  #                                  # lead you opened by hand is not relaunched under you.
-  #     tabPrefix: "lead:"           # `lead: opus-5.0` ⇒ a lead named opus-5.0 (case-insensitive)
+  #     tab: "lead: opus-5.0"        # REQUIRED — the exact tab label this lead lives in
+  #     tabPrefix: "lead:"           # only used to guard against a worker tabLabel colliding with
+  #                                  # this convention at startup; plays no part in matching a lead
  #     scanIntervalSeconds: 10      # rescan cadence, and the worst case before a new tab is seen
  #     workspace: leads             # where a launched lead's tab is created (default "leads").
  #                                  # MUST NOT be a member workspace — those are excluded from the
@@ -337,7 +349,7 @@ fleet:
  #     cwd: /path/to/repo           # the launched lead's working directory (default: bridged's own)
  #     kind: claude                 # descriptive; reported by bridge_whoami
  #   gpt-sol-5.6:
-  #     terminal: term_fedcba9876543
+  #     tab: "lead: gpt-sol-5.6"
  #     kind: opencode
  #     model: openai/gpt-5.6-terra

@@ -13,6 +13,7 @@ import dev.ltms.bridged.herdr.PaneLocator;
 import dev.ltms.bridged.herdr.UnixSocketHerdrClient;
 import dev.ltms.bridged.herdr.WorkspaceControl;
 import dev.ltms.bridged.inject.CompletionResolver;
+import dev.ltms.bridged.inject.ExhaustedPatternLookup;
 import dev.ltms.bridged.inject.Injector;
 import dev.ltms.bridged.inject.StatusPoller;
 import dev.ltms.bridged.inject.TurnListener;
@@ -55,12 +56,12 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;

 /**
@@ -194,34 +195,34 @@ public final class Bridged {
        if (leadTerminals.size() > 1) {
            log.info("leads: {} panes recognised {}", leadTerminals.size(), leadTerminals.values());
        }
-        // CB-531: on top of the static registry, discover leads by the tab labels the operator
-        // writes. CB-557 moved the settings onto the lead they describe, so scanning is on whenever
-        // a `fleet.leaders:` entry exists — with no leads configured the supplier is a constant and
-        // never touches herdr, exactly as a missing `leadScan:` block used to behave.
+        // CB-531: on top of the legacy primary.terminal pin, discover leads by the tab labels the
+        // operator writes. CB-557 moved the settings onto the lead they describe, so scanning is on
+        // whenever a `fleet.leaders:` entry exists — with no leads configured the supplier is a
+        // constant and never touches herdr, exactly as a missing `leadScan:` block used to behave.
+        // CB-579: each lead now names its own exact `tab:` label, so one scanner discovers every
+        // configured lead regardless of how differently their tabs are labelled — the old
+        // single-shared-tabPrefix limitation (and its warning) is gone.
        final Supplier<Map<String, String>> leads;
        var leaders = cfg.fleet().leaders();
        if (!leaders.isEmpty()) {
-            // One scanner, so one prefix and one interval. Distinct per-lead prefixes would need a
-            // scanner each; until a config actually wants that, take the first entry's settings and
-            // say so, rather than silently honouring one lead's prefix and dropping another's.
-            var scan = leaders.values().iterator().next();
            Set<String> memberSpaces = cfg.profiles().values().stream()
                    .map(BridgedConfig.Profile::workspace)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toSet());
-            leads = new LeadTabScanner(herdr, scan.tabPrefix(), memberSpaces, leadTerminals,
-                    TimeUnit.SECONDS.toNanos(scan.scanIntervalSeconds()), System::nanoTime);
-            log.info("lead scan: tabs labelled '{}…' host a lead (rescan every {}s, member spaces {} "
-                            + "excluded)",
-                    scan.tabPrefix(), scan.scanIntervalSeconds(), memberSpaces);
-            long distinctPrefixes = leaders.values().stream()
-                    .map(BridgedConfig.Leader::tabPrefix).distinct().count();
-            if (distinctPrefixes > 1) {
-                log.warn("fleet.leaders declares {} different tabPrefix values; only '{}' is scanned "
-                                + "for. Give every lead the same tabPrefix, or leads under the others "
-                                + "will not be discovered.",
-                        distinctPrefixes, scan.tabPrefix());
-            }
+            Map<String, String> tabToName = new LinkedHashMap<>();
+            leaders.forEach((name, leader) -> {
+                if (leader != null && leader.tab() != null && !leader.tab().isBlank()) {
+                    tabToName.put(leader.tab(), name);
+                }
+            });
+            // One shared rescan cadence: still taken from the first entry, as before — it is an
+            // operational cadence, not identity, so there is no correctness reason to give every
+            // lead its own scanner.
+            int scanIntervalSeconds = leaders.values().iterator().next().scanIntervalSeconds();
+            leads = new LeadTabScanner(herdr, tabToName, memberSpaces,
+                    TimeUnit.SECONDS.toNanos(scanIntervalSeconds), System::nanoTime);
+            log.info("lead scan: tabs {} host a lead (rescan every {}s, member spaces {} excluded)",
+                    tabToName.keySet(), scanIntervalSeconds, memberSpaces);
        } else {
            leads = () -> leadTerminals;
        }
@@ -254,8 +255,24 @@ public final class Bridged {
        // The blocking message endpoint (CB-104) is the producer; the poller is inert until then.
        // CB-106: a confirmed turn completion resolves a blocked send whose worker never replied.
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver completion = new CompletionResolver(agents, rendezvous);
-        AtomicReference<MessageService> messagesRef = new AtomicReference<>();
+        // CB-578 stage A: classify a completion-fallback scrape that matches a profile's configured
+        // usage-limit refusal as BACKEND_EXHAUSTED rather than handing it back as a real answer.
+        // Compiled once at startup, keyed by profile name; a profile with no exhaustedPattern is
+        // simply absent here, so its workers keep today's completion-fallback behaviour unchanged.
+        Map<String, Pattern> exhaustedPatternsByProfile = new LinkedHashMap<>();
+        cfg.profiles().forEach((name, profile) -> {
+            if (profile.hasExhaustedPattern()) {
+                exhaustedPatternsByProfile.put(name, Pattern.compile(profile.exhaustedPattern()));
+            }
+        });
+        ExhaustedPatternLookup exhaustedPatterns = target -> sessions.roster().stream()
+                .filter(session -> target.equals(session.terminalId()))
+                .findFirst()
+                .map(session -> exhaustedPatternsByProfile.get(session.profile()))
+                .orElse(null);
+        log.info("backend-exhausted classification (CB-578 stage A): {}",
+                CompletionResolver.coverage(cfg.profiles().keySet(), exhaustedPatternsByProfile.keySet()));
+        CompletionResolver completion = new CompletionResolver(agents, rendezvous, exhaustedPatterns);
        // CB-113: deliver only to an available worker (its MCP is connected), never its boot window.
        // CB-301: the manager's presence bridge records availability and drives SPAWNING → READY.
        MemberPresence presence = sessions.asPresence();
@@ -278,23 +295,21 @@ public final class Bridged {
            }

            @Override
-            public void onDelivered(String target) {
-                completion.onDelivered(target);
-                sessions.onDelivered(target);
+            public void onDelivered(String target, dev.ltms.bridged.msg.TurnToken token) {
+                completion.onDelivered(target, token);
+                sessions.onDelivered(target, token);
            }

            @Override
            public void onTurnFailed(String target) {
                completion.onTurnFailed(target);
                sessions.onTurnFailed(target);
-                failTarget(messagesRef, target, "worker turn failed");
            }

            @Override
            public void onTurnFailed(String target, String reason) {
                completion.onTurnFailed(target, reason);
                sessions.onTurnFailed(target);
-                failTarget(messagesRef, target, reason);
            }
        };
        Injector injector = new Injector(agents, turnListener, deliverableTo(presence, leads),
@@ -358,13 +373,14 @@ public final class Bridged {
        }
        MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
                pushLoop, metrics);
-        messagesRef.set(messages);

        // Health is a slow whole-fleet observer. Keep it separate from the 250ms delivery poller.
        final FleetHealthMonitor healthMonitor;
        var healthScheduler = Executors.newSingleThreadScheduledExecutor(r ->
                Thread.ofVirtual().name("bridge-health-").unstarted(r));
        if (cfg.health() != null && cfg.health().isEnabled()) {
+            // CB-580: a member found GONE/NEVER_READY must fail whatever ticket is waiting on it,
+            // through the same idempotent target-wide operation CB-516 already uses on release.
            healthMonitor = new FleetHealthMonitor(agents, sessions::roster, messages, healthScheduler,
                    System::nanoTime, cfg.health().intervalOrDefault(), messages::abandon);
            String coverage = FleetHealthMonitor.coverage(true,
@@ -490,14 +506,6 @@ public final class Bridged {
        return target -> presence.isPresent(target) || leads.get().containsKey(target);
    }

-    /** Fail outstanding tickets without changing the member lifecycle or worktree state. */
-    private static void failTarget(AtomicReference<MessageService> messagesRef, String target, String reason) {
-        MessageService messages = messagesRef.get();
-        if (messages != null) {
-            messages.abandon(target, reason);
-        }
-    }
-
    /**
     * Poll herdr's {@code ping} until it answers or {@link #HERDR_WAIT_SECONDS} elapses (CB-504).
     *
@@ -181,6 +181,12 @@ public record BridgedConfig(
     *                       opposite intents). For the same reason, an {@code env:} entry naming
     *                       {@code ANTHROPIC_BASE_URL} or {@code ANTHROPIC_AUTH_TOKEN} is refused at
     *                       config load (CB-542): on the subscription path no guard would vet it.
+     * @param exhaustedPattern  regex matched against a completion-fallback scrape (CB-578 stage A) to
+     *                       classify a turn that ended with no {@code bridge_reply} as the backend
+     *                       having refused on a subscription usage limit, rather than a real answer.
+     *                       {@code null}/blank ⇒ the classification never fires for this profile and
+     *                       today's completion-fallback behaviour is unchanged. Every backend words
+     *                       its refusal differently, so this is config, never a vendor string in code.
     */
    @JsonIgnoreProperties(ignoreUnknown = true)
    public record Profile(String profile, String baseUrl, String model,
@@ -193,7 +199,8 @@ public record BridgedConfig(
                         Map<String, String> env,
                         Float weight,
                         Integer maxLoad,
-                         Boolean subscription) {
+                         Boolean subscription,
+                         String exhaustedPattern) {

        /** Peer kind spawned by {@link dev.ltms.bridged.member.ClaudeCodeLauncher} (the default). */
        public static final String KIND_CLAUDE_CODE = "claude-code";
@@ -236,6 +243,9 @@ public record BridgedConfig(
            weight = (weight == null || weight <= 0.0f) ? 1.0f : weight;
            maxLoad = (maxLoad == null || maxLoad <= 0) ? null : maxLoad;
            subscription = (subscription != null && subscription) ? Boolean.TRUE : Boolean.FALSE;
+            // exhaustedPattern stays null when unset/blank (opt-in) — no defaulting, no vendor
+            // wording: an unconfigured profile keeps today's completion-fallback behaviour exactly.
+            exhaustedPattern = (exhaustedPattern == null || exhaustedPattern.isBlank()) ? null : exhaustedPattern;
        }

        /**
@@ -248,7 +258,7 @@ public record BridgedConfig(
                      String placement, String workspace, String tabLabel, String mcpUrl,
                      String cwd, List<String> parityOverlay) {
            this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
-                    mcpUrl, cwd, parityOverlay, null, null, null, null, null, null, null);
+                    mcpUrl, cwd, parityOverlay, null, null, null, null, null, null, null, null);
        }

        /**
@@ -260,7 +270,7 @@ public record BridgedConfig(
                      String placement, String workspace, String tabLabel, String mcpUrl,
                      String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv) {
            this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
-                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, null, null, null, null, null);
+                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, null, null, null, null, null, null);
        }

        /**
@@ -273,13 +283,14 @@ public record BridgedConfig(
                      String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv,
                      String kind) {
            this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
-                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, null, null, null, null);
+                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, null, null, null, null, null);
        }

        /** A copy with {@code profile} set — used to default a profile to its {@code workers} key. */
        public Profile withProfile(String p) {
            return new Profile(p, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
-                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, subscription);
+                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, subscription,
+                    exhaustedPattern);
        }

        /** True when this profile is served by the Claude Code adapter (the default kind). */
@@ -312,7 +323,12 @@ public record BridgedConfig(
                      String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv,
                      String kind, Map<String, String> env, Float weight, Integer maxLoad) {
            this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
-                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, null);
+                    mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, null, null);
+        }
+
+        /** True when this profile's CB-578 stage A backend-exhausted classification is configured. */
+        public boolean hasExhaustedPattern() {
+            return exhaustedPattern != null;
        }

        /** True when this profile's workers are granted a forge token to open their own PR (CB-302). */
@@ -458,24 +474,32 @@ public record BridgedConfig(
     * pre-existed, which is why it had to be recognised by configuration rather than created. With
     * {@code profile} and {@code instances} the daemon may stand one up when none is live, so the
     * pane no longer has to exist before the daemon does. Recognition still comes first: a lead
-     * already running under {@code tabPrefix} is adopted, and only the shortfall is launched.
+     * already running in its configured {@code tab} is adopted, and only the shortfall is launched.
+     *
+     * <p><b>{@code tab} replaced {@code terminal} (CB-579).</b> A herdr {@code terminal_id} changes
+     * every time the lead's session restarts, so pinning one cost a config edit and a daemon restart
+     * per restart. A tab is stable: a human opens it once, it holds exactly one pane, and its label
+     * survives restarts of the agent inside it — so identity is now the tab label alone.
     *
     * @param profile             the {@code profiles:} entry to launch this lead on when one must
     *                            be created; {@code null} ⇒ recognise-only, never create
-     * @param terminal            the lead's herdr {@code terminal_id} when pinned by hand; the only
-     *                            field identity depends on. {@code null} ⇒ found by {@code tabPrefix}
+     * @param tab                 the exact tab label hosting this lead, matched case-insensitively;
+     *                            the only field identity depends on. Required — a lead with no
+     *                            {@code tab} can never be discovered, launched or not
     * @param instances           how many of this lead should be live (default 1). The daemon
     *                            launches only the shortfall, so a restart adopts rather than doubles
-     * @param tabPrefix           label prefix marking this lead's tab, matched case-insensitively;
-     *                            the remainder is the lead's name ({@code "lead: opus"} →
-     *                            {@code opus}). Default {@code "lead:"}
+     * @param tabPrefix           no longer used to find a lead's tab — {@code tab} is matched
+     *                            exactly. Its only remaining job is the startup collision guard
+     *                            ({@link #validateLeadTabPrefixes()}), which still uses it to refuse
+     *                            a worker {@code tabLabel} template that could be misread as a lead.
+     *                            Default {@code "lead:"}
     * @param scanIntervalSeconds how long a tab scan is cached before herdr is asked again; also the
     *                            worst case before a newly-labelled tab is recognised. Default 10
     * @param kind                which agent runs there ({@code claude}, {@code opencode}, …)
     * @param model               the model or selector it runs, for operators reading the roster
     */
    @JsonIgnoreProperties(ignoreUnknown = true)
-    public record Leader(String profile, String terminal, Integer instances, String tabPrefix,
+    public record Leader(String profile, String tab, Integer instances, String tabPrefix,
                         Integer scanIntervalSeconds, String kind, String model,
                         String workspace, String cwd) {

@@ -493,12 +517,13 @@ public record BridgedConfig(
                    (scanIntervalSeconds == null || scanIntervalSeconds <= 0) ? 10 : scanIntervalSeconds;
            workspace = (workspace == null || workspace.isBlank())
                    ? DEFAULT_WORKSPACE : workspace.strip();
+            tab = (tab == null || tab.isBlank()) ? null : tab.strip();
        }

        /** Back-compat 7-arg form — no workspace or cwd, so both take their defaults. */
-        public Leader(String profile, String terminal, Integer instances, String tabPrefix,
+        public Leader(String profile, String tab, Integer instances, String tabPrefix,
                      Integer scanIntervalSeconds, String kind, String model) {
-            this(profile, terminal, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
+            this(profile, tab, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
        }

        /** True when this lead may be launched by the daemon rather than only recognised. */
@@ -506,9 +531,9 @@ public record BridgedConfig(
            return profile != null && !profile.isBlank() && instances > 0;
        }

-        /** The tab label an auto-launched instance of this lead gets — what the scanner reads back. */
-        public String tabLabel(String name) {
-            return tabPrefix + " " + name;
+        /** The tab label an auto-launched instance of this lead gets — its configured {@code tab}. */
+        public String tabLabel() {
+            return tab;
        }
    }

@@ -707,28 +732,20 @@ public record BridgedConfig(
    }

    /**
-     * The terminal → lead-name map that {@link dev.ltms.bridged.auth.CallerResolver} resolves
-     * against, merging the {@code leaders:} registry with the legacy singular {@code primary:} pin.
+     * The terminal → lead-name map seeded from the legacy singular {@code primary:} pin (CB-530).
     *
-     * <p>Precedence: an explicit {@code leaders:} entry wins over the {@code primary:} pin for the
-     * same terminal. The pin is the older, less expressive spelling of the same fact, so when both
-     * name a pane the named entry is the one an operator meant. The pin is still honoured on its
-     * own — a config carrying only {@code primary:} behaves exactly as it did before CB-530.
+     * <p>{@code fleet.leaders} no longer carries a per-entry terminal pin (CB-579): a lead's identity
+     * comes from its {@code tab} alone, resolved live by {@code LeadTabScanner}. This method now
+     * exists only for the {@code primary.terminal} fallback — a config that never migrated off it
+     * still resolves that one pane as a lead named {@code "primary"}, exactly as before CB-530.
     *
-     * @return an unmodifiable map, empty when neither block is configured (nothing is pinned, and
-     *         every pane therefore resolves as a worker — the pre-CB-307 behaviour)
+     * @return an unmodifiable map, empty when {@code primary.terminal} is not configured (nothing is
+     *         pinned, and every pane therefore resolves as a worker — the pre-CB-307 behaviour)
     */
    public Map<String, String> leaderTerminals() {
        Map<String, String> byTerminal = new LinkedHashMap<>();
-        if (fleet != null) {
-            fleet.leaders().forEach((name, leader) -> {
-                if (leader != null && leader.terminal() != null && !leader.terminal().isBlank()) {
-                    byTerminal.put(leader.terminal(), name);
-                }
-            });
-        }
        if (primary != null && primary.terminal() != null && !primary.terminal().isBlank()) {
-            byTerminal.putIfAbsent(primary.terminal(), "primary");
+            byTerminal.put(primary.terminal(), "primary");
        }
        return Collections.unmodifiableMap(byTerminal);
    }
@@ -840,6 +857,7 @@ public record BridgedConfig(
        try {
            String yaml = Files.readString(path);
            rejectRenamedTopLevelKeys(yaml);
+            rejectLeaderTerminalKey(yaml);
            warnUnknownTopLevelKeys(yaml, path);
            rejectDuplicateMemberSlots(yaml);
            BridgedConfig cfg = YAML.readValue(yaml, BridgedConfig.class);
@@ -1064,6 +1082,43 @@ public record BridgedConfig(
        }
    }

+    /**
+     * Reject a config whose {@code fleet.leaders.<name>} still carries the retired {@code terminal:}
+     * pin (CB-579), naming {@code tab:} as its replacement.
+     *
+     * <p>{@code Leader} is {@code @JsonIgnoreProperties(ignoreUnknown = true)}, so simply dropping
+     * the record component would make a leftover {@code terminal:} key silently no-op — the daemon
+     * would start, the pin would never take effect, and nothing would say why. Fatal and specific
+     * instead, exactly like {@link #rejectRenamedTopLevelKeys}, which this mirrors for a key one
+     * level deeper than the ones that method covers.
+     *
+     * @param yaml the raw config text
+     * @throws IllegalStateException when any {@code fleet.leaders.<name>.terminal} key is present
+     */
+    static void rejectLeaderTerminalKey(String yaml) {
+        Map<?, ?> raw;
+        try {
+            raw = YAML.readValue(yaml, Map.class);
+        } catch (IOException | IllegalArgumentException e) {
+            return; // a malformed file is reported by the real parse, not here
+        }
+        if (raw == null || !(raw.get("fleet") instanceof Map<?, ?> fleet)
+                || !(fleet.get("leaders") instanceof Map<?, ?> leaders)) {
+            return;
+        }
+        List<String> bad = leaders.entrySet().stream()
+                .filter(e -> e.getValue() instanceof Map<?, ?> leader && leader.containsKey("terminal"))
+                .map(e -> String.valueOf(e.getKey()))
+                .sorted()
+                .toList();
+        if (!bad.isEmpty()) {
+            throw new IllegalStateException("refusing to start: fleet.leaders entries ["
+                    + String.join(", ", bad) + "] still use the retired 'terminal:' key — replace it "
+                    + "with 'tab:', the exact tab label hosting the lead. A terminal_id changes on "
+                    + "every restart of the lead's session; a tab label does not.");
+        }
+    }
+
    static List<String> unknownTopLevelKeys(String yaml) {
        Map<?, ?> raw;
        try {
@@ -1313,10 +1368,9 @@ public record BridgedConfig(
                        + "', which is not a configured profiles: entry (have: " + profiles.keySet()
                        + ").");
            }
-            if (!leader.isCreatable() && (leader.terminal() == null || leader.terminal().isBlank())) {
-                bad.add("fleet.leaders." + name + " can neither be found nor created — it pins no "
-                        + "terminal: and names no profile: to launch one on. Give it one or the "
-                        + "other, or drop the entry.");
+            if (leader.tab() == null || leader.tab().isBlank()) {
+                bad.add("fleet.leaders." + name + " has no tab: — a lead is now found (and, if "
+                        + "auto-launched, labelled) purely by its tab, so every entry must name one.");
            }
        });
        if (!bad.isEmpty()) {
@@ -30,7 +30,11 @@ import java.util.function.Supplier;
 *       {@code fleet:} (every role pool, {@code charters}, and {@code tabLabel}),
 *       {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Those
 *       three are read through a supplier on {@code CompositePeerLauncher}, which is what makes
- *       them hot — not the fact that they are config.</li>
+ *       them hot — not the fact that they are config. <strong>This does NOT include
+ *       {@code fleet.leaders}</strong>: {@code Bridged.main} reads {@code cfg.fleet().leaders()}
+ *       once at startup to build the {@code LeadTabScanner} and the {@code LeadLauncher}, and
+ *       neither is reconstructed on reload — so a lead added, removed, or re-{@code tab}'d under
+ *       {@code fleet.leaders} needs a restart, the same as any deferred key below.</li>
 *   <li><strong>Deferred</strong> — accepted into the new snapshot, but the wiring built at startup
 *       keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:},
 *       {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code guard:},
@@ -12,15 +12,20 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
-import java.util.function.BiConsumer;

 /** Slow whole-fleet evidence collection. It is deliberately separate from the delivery poller. */
 public final class FleetHealthMonitor {
    private static final Logger log = LoggerFactory.getLogger(FleetHealthMonitor.class);
+
+    /** Bounded attempts to run {@link #failTarget} for one transition. Never retried tick-to-tick (CB-580). */
+    static final int MAX_FAIL_TARGET_ATTEMPTS = 3;
+
    private final AgentControl agents;
    private final Supplier<List<MemberSession>> roster;
    private final MessageService messages;
@@ -34,21 +39,23 @@ public final class FleetHealthMonitor {
    // These facts need the evidence publishers introduced by later M4 units. They are not negatives.
    private static final boolean NOT_YET_OBSERVED = false;

+    /**
+     * @param failTarget CB-568's idempotent target-wide failure operation (e.g. {@code messages::abandon}),
+     *                   invoked once when a member transitions into a terminal health state. Required —
+     *                   there is deliberately no defaulting overload; a caller that does not want the
+     *                   fail-tickets-on-terminal-health behavior must pass an explicit inert value (see
+     *                   {@code TestTurnTokens.inert} / {@code BridgeMcp.CapacitySource.none()} for the pattern).
+     */
    public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
-                               ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds) {
-        this(agents, roster, messages, scheduler, clock, intervalSeconds, (_, _) -> { });
-    }
-
-    public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
-                               ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds,
-                               BiConsumer<String, String> failTarget) {
+                              ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds,
+                              BiConsumer<String, String> failTarget) {
        this.agents = agents;
        this.roster = roster;
        this.messages = messages;
        this.scheduler = scheduler;
        this.clock = clock;
        this.intervalSeconds = intervalSeconds;
-        this.failTarget = failTarget;
+        this.failTarget = Objects.requireNonNull(failTarget, "failTarget");
    }

    /** Pure per-member decision seam. */
@@ -78,9 +85,6 @@ public final class FleetHealthMonitor {
                HealthDecision decision = decide(snapshot, priors.getOrDefault(session.terminalId(), HealthPrior.NONE),
                        clock.getAsLong());
                priors.put(session.terminalId(), decision.prior());
-                if (decision.state() == HealthState.GONE || decision.state() == HealthState.NEVER_READY) {
-                    failTarget.accept(session.terminalId(), "fleet health detected " + decision.state());
-                }
                reportTransition(session.terminalId(), decision.state());
            }
            priors.keySet().retainAll(current);
@@ -103,6 +107,34 @@ public final class FleetHealthMonitor {
        } else if (previous != null && fault(previous)) {
            log.info("fleet health member={} recovered state={} previous={}", target, next, previous);
        }
+        // CB-580: a member entering GONE/NEVER_READY must not leave its waiting tickets pending
+        // forever. Fire exactly once per transition — never on a tick where the state is unchanged,
+        // which is what made the rejected commit call abandon() once per tick for as long as a
+        // member stayed terminal.
+        if (terminal(next)) {
+            failTerminalTarget(target, next);
+        }
+    }
+
+    private void failTerminalTarget(String target, HealthState state) {
+        String reason = "fleet health: member reached terminal state " + state.name();
+        RuntimeException last = null;
+        for (int attempt = 1; attempt <= MAX_FAIL_TARGET_ATTEMPTS; attempt++) {
+            try {
+                failTarget.accept(target, reason);
+                return;
+            } catch (RuntimeException error) {
+                last = error;
+                log.warn("fleet health: failTarget attempt {}/{} failed for member={} state={}",
+                        attempt, MAX_FAIL_TARGET_ATTEMPTS, target, state, error);
+            }
+        }
+        log.warn("fleet health: giving up on failTarget for member={} state={} after {} attempts",
+                target, state, MAX_FAIL_TARGET_ATTEMPTS, last);
+    }
+
+    private static boolean terminal(HealthState state) {
+        return state == HealthState.GONE || state == HealthState.NEVER_READY;
    }

    private static boolean fault(HealthState state) {
@@ -6,6 +6,7 @@ import org.slf4j.LoggerFactory;

 import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.LongSupplier;
@@ -23,6 +24,15 @@ import java.util.function.Supplier;
 * by first starting the session and asking it. Scanning closes that loop: label the tab, and the
 * pane is recognised on the next resolve.
 *
+ * <p><strong>CB-579 — matched by name, not prefix.</strong> This used to strip one shared
+ * {@code tabPrefix} off a label to derive the lead's name, and merged a config-supplied
+ * {@code terminal_id} pin over every scan result so the pin could never expire. Both are gone: each
+ * lead now configures its own exact {@code tab} label ({@code fleet.leaders.<name>.tab}), so this
+ * class is handed a {@code tab → name} map up front and matches labels against it exactly
+ * (case-insensitively). There is no merge step — a scan result is the whole answer. That is the
+ * fix for the bug this replaces: a {@code terminal_id} pin surviving in config after the pane it
+ * named was gone, so the daemon kept treating a dead session as a live lead forever.
+ *
 * <p><strong>Direction of trust.</strong> The label names the lead; it never <em>grants</em>
 * anything a pane could take for itself. Three properties keep that honest:
 * <ol>
@@ -48,7 +58,7 @@ import java.util.function.Supplier;
 * 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 BridgedConfig.validateLeadScan} rather than documented here.
+ * startup by {@code BridgedConfig.validateLeadTabPrefixes} rather than documented here.
 *
 * <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
@@ -60,38 +70,46 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
    private static final Logger log = LoggerFactory.getLogger(LeadTabScanner.class);

    private final HerdrClient herdr;
-    private final String tabPrefix;
+    private final Map<String, String> tabToName;
    private final Set<String> excludedWorkspaceLabels;
-    private final Map<String, String> configuredLeads;
    private final long ttlNanos;
    private final LongSupplier clock;

-    private Map<String, String> cached;
+    private Map<String, String> cached = Map.of();
    private long scannedAtNanos;
    private boolean everScanned;

    /**
     * @param herdr                   the herdr client to query ({@code workspace.list},
     *                                {@code tab.list}, {@code pane.list} — all read-only)
-     * @param tabPrefix               a tab whose label starts with this (case-insensitively) hosts a
-     *                                lead; the rest of the label, trimmed, is the lead's name
+     * @param tabToName               every configured lead's exact tab label → its name
+     *                                ({@code fleet.leaders.<name>.tab}), matched case-insensitively
     * @param excludedWorkspaceLabels workspaces never scanned — the configured worker spaces
-     * @param configuredLeads         the static {@code leaders:}/{@code primary:} registry, merged
-     *                                over every scan result. Explicit config outranks the
-     *                                convention, and survives a scan that cannot run at all
     * @param ttlNanos                how long a scan result is reused before the next one
     * @param clock                   nanosecond time source ({@code System::nanoTime} in production)
     */
-    public LeadTabScanner(HerdrClient herdr, String tabPrefix, Set<String> excludedWorkspaceLabels,
-                          Map<String, String> configuredLeads, long ttlNanos, LongSupplier clock) {
+    public LeadTabScanner(HerdrClient herdr, Map<String, String> tabToName,
+                          Set<String> excludedWorkspaceLabels, long ttlNanos, LongSupplier clock) {
        this.herdr = herdr;
-        this.tabPrefix = tabPrefix == null || tabPrefix.isBlank() ? "lead:" : tabPrefix.strip();
+        this.tabToName = normalize(tabToName);
        this.excludedWorkspaceLabels = excludedWorkspaceLabels == null
                ? Set.of() : Set.copyOf(excludedWorkspaceLabels);
-        this.configuredLeads = configuredLeads == null ? Map.of() : Map.copyOf(configuredLeads);
        this.ttlNanos = ttlNanos;
        this.clock = clock;
-        this.cached = this.configuredLeads;
+    }
+
+    /** Keys stripped and lower-cased once, so every lookup is a plain map hit. */
+    private static Map<String, String> normalize(Map<String, String> tabToName) {
+        if (tabToName == null || tabToName.isEmpty()) {
+            return Map.of();
+        }
+        Map<String, String> out = new LinkedHashMap<>();
+        tabToName.forEach((tab, name) -> {
+            if (tab != null && !tab.isBlank() && name != null && !name.isBlank()) {
+                out.put(tab.strip().toLowerCase(Locale.ROOT), name);
+            }
+        });
+        return Collections.unmodifiableMap(out);
    }

    /**
@@ -151,25 +169,20 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
                }
            }
        }
-        byTerminal.putAll(configuredLeads); // an explicit pin outranks a label
        return Collections.unmodifiableMap(byTerminal);
    }

    /**
-     * The lead name a tab label declares, or {@code null} if it declares none.
+     * The lead name a tab label declares, or {@code null} if it names none of the configured leads.
     *
-     * <p>{@code "lead: opus-5.0"} → {@code "opus-5.0"}. A bare {@code "lead:"} names nobody and is
-     * rejected: an unnamed lead would resolve as {@code PRIMARY} with nothing to attribute it to.
+     * <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.
     */
    private String leadNameOf(String label) {
        if (label == null) {
            return null;
        }
-        String l = label.strip();
-        if (!l.regionMatches(true, 0, tabPrefix, 0, tabPrefix.length())) {
-            return null;
-        }
-        String name = l.substring(tabPrefix.length()).strip();
-        return name.isEmpty() ? null : name;
+        return tabToName.get(label.strip().toLowerCase(Locale.ROOT));
    }
 }
@@ -2,11 +2,17 @@ package dev.ltms.bridged.inject;

 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;

 /**
 * The CB-106 completion fallback: bridges the {@link Injector}'s turn-completion signal to the
@@ -60,6 +66,7 @@ public final class CompletionResolver implements TurnListener {

    private final AgentControl agents;
    private final Rendezvous rendezvous;
+    private final ExhaustedPatternLookup exhaustedPatterns;

    /**
     * Per-target record of the turn currently in flight: the exact {@link Rendezvous} waiter its
@@ -77,23 +84,30 @@ public final class CompletionResolver implements TurnListener {

    private final ConcurrentHashMap<String, InFlight> inFlight = new ConcurrentHashMap<>();

-    public CompletionResolver(AgentControl agents, Rendezvous rendezvous) {
+    /**
+     * @param exhaustedPatterns CB-578 stage A: per-target lookup for a profile's configured
+     *                          usage-limit refusal pattern. Required — there is deliberately no
+     *                          defaulting overload; a caller that does not want the classification
+     *                          must pass an explicit inert value ({@link ExhaustedPatternLookup#none()}).
+     */
+    public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns) {
        this.agents = agents;
        this.rendezvous = rendezvous;
+        this.exhaustedPatterns = Objects.requireNonNull(exhaustedPatterns, "exhaustedPatterns");
    }

    @Override
-    public void onDelivered(String target) {
+    public void onDelivered(String target, TurnToken token) {
        // Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn
        // content — what it shows *before* the just-delivered turn produces output — as the staleness
        // reference (CB-115). Done synchronously (like the delivering send itself) so both are in
        // place before this turn's completion can fire.
-        captureBaseline(target);
+        captureBaseline(target, token);
    }

    /** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
-    void captureBaseline(String target) {
-        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
+    void captureBaseline(String target, TurnToken token) {
+        CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
        if (waiter == null) {
            inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later
            return;
@@ -156,11 +170,12 @@ public final class CompletionResolver implements TurnListener {
            return;
        }
        String tail;
+        String assistantBlock = null;
        int originalLength = 0;
        boolean clipped = false;
        boolean scrapeFailed = false;
        try {
-            String assistantBlock = lastAssistantBlock(agents.read(target, SCRAPE_SOURCE));
+            assistantBlock = lastAssistantBlock(agents.read(target, SCRAPE_SOURCE));
            originalLength = assistantBlock.strip().length();
            clipped = originalLength > MAX_SCRAPE_CHARS;
            tail = clip(assistantBlock);
@@ -183,6 +198,22 @@ public final class CompletionResolver implements TurnListener {
                    target);
            return; // keep the in-flight record: a later genuine completion still needs it
        }
+        // CB-578 stage A: a turn that ended with no bridge_reply AND whose scrape matches the
+        // backend's configured usage-limit pattern is a refusal, not an answer. Classify it as
+        // BACKEND_EXHAUSTED rather than handing the caller a scrape that reads like a real reply.
+        if (!scrapeFailed) {
+            Pattern exhausted = exhaustedPatterns.patternFor(target);
+            String matchedLine = exhausted == null ? null : firstMatchingLine(assistantBlock, exhausted);
+            if (matchedLine != null) {
+                String reason = "backend exhausted (usage limit): " + matchedLine;
+                if (rendezvous.resolveExhausted(waiter, reason)) {
+                    inFlight.remove(target, turn);
+                    log.warn("completion for {} classified BACKEND_EXHAUSTED (no bridge_reply; scrape "
+                            + "matched the profile's exhausted pattern): {}", target, reason);
+                }
+                return;
+            }
+        }
        String completion = clipped ? tail + "\n" + CLIPPED_PANE_TAIL_MARKER : tail;
        if (rendezvous.resolveCompletion(waiter, completion)) {
            inFlight.remove(target, turn);
@@ -231,6 +262,45 @@ public final class CompletionResolver implements TurnListener {
        }
    }

+    /**
+     * The first line of {@code text} matching {@code pattern}, stripped — the CB-578 stage A
+     * evidence carried in a {@code BACKEND_EXHAUSTED} reason so the operator sees the real refusal
+     * text, never a generic label. {@code null} if no line matches.
+     */
+    static String firstMatchingLine(String text, Pattern pattern) {
+        if (text == null || text.isEmpty()) return null;
+        for (String line : text.split("\n", -1)) {
+            if (pattern.matcher(line).find()) {
+                return line.strip();
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Coverage summary for the CB-578 stage A exhausted-pattern classification, logged at startup
+     * the way {@link dev.ltms.bridged.health.FleetHealthMonitor#coverage} is — so an operator can
+     * see whether the classification is on, and for which profiles, without reading every
+     * profile's config by hand.
+     *
+     * @param allProfiles        every configured profile name
+     * @param configuredProfiles the subset of {@code allProfiles} that carry an exhausted pattern
+     */
+    public static String coverage(Set<String> allProfiles, Set<String> configuredProfiles) {
+        if (configuredProfiles.isEmpty()) {
+            return "off (no profile has an exhaustedPattern configured; profiles: " + sorted(allProfiles) + ")";
+        }
+        Set<String> unconfigured = new TreeSet<>(allProfiles);
+        unconfigured.removeAll(configuredProfiles);
+        return unconfigured.isEmpty()
+                ? "full (all profiles configured: " + sorted(allProfiles) + ")"
+                : "partial (configured: " + sorted(configuredProfiles) + "; not configured: " + sorted(unconfigured) + ")";
+    }
+
+    private static List<String> sorted(Set<String> names) {
+        return names.stream().sorted().toList();
+    }
+
    private static String clip(String s) {
        if (s == null) return "";
        String trimmed = s.strip();
@@ -0,0 +1,27 @@
+package dev.ltms.bridged.inject;
+
+import java.util.regex.Pattern;
+
+/**
+ * Per-target lookup for a profile's configured usage-limit refusal pattern (CB-578 stage A): how
+ * {@link CompletionResolver} tells a backend that refused on a subscription usage limit — the
+ * worker's pane stays healthy, but the account is exhausted — apart from a genuine completion.
+ *
+ * <p>The pattern is always profile config, never a vendor string in Java source: every backend
+ * words its refusal differently, so a hardcoded sentence would only ever match one of them.
+ */
+@FunctionalInterface
+public interface ExhaustedPatternLookup {
+
+    /** The compiled pattern configured for {@code target}'s profile, or {@code null} if none. */
+    Pattern patternFor(String target);
+
+    /**
+     * Inert lookup — no profile has a pattern configured, so the classification never fires and
+     * the completion fallback behaves exactly as before CB-578 stage A. The explicit stand-in a
+     * caller (or a test not exercising this feature) passes instead of a defaulting overload.
+     */
+    static ExhaustedPatternLookup none() {
+        return target -> null;
+    }
+}
@@ -2,6 +2,7 @@ package dev.ltms.bridged.inject;

 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

@@ -129,7 +130,7 @@ public final class Injector {
    }

    /** A pending message and the future that completes when it has been delivered. */
-    private record Pending(String text, CompletableFuture<Void> delivered) {
+    private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
    }

    /** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -159,9 +160,9 @@ public final class Injector {
     * <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
     * target" and "queue the message" and orphan it in a target it just removed.
     */
-    public CompletableFuture<Void> enqueue(String target, String text) {
+    public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
        CompletableFuture<Void> delivered = new CompletableFuture<>();
-        Pending p = new Pending(text, delivered);
+        Pending p = new Pending(text, token, delivered);
        targets.compute(target, (_, existing) -> {
            Target t = (existing != null) ? existing : new Target();
            t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
@@ -356,7 +357,7 @@ public final class Injector {
            } else {
                // Baseline the pane's pre-turn content so a misattributed completion (no new output)
                // can't resolve this send with the previous turn's stale answer (CB-115).
-                turnListener.onDelivered(target);
+                turnListener.onDelivered(target, sent.token());
                sent.delivered().complete(null);
            }
        }
@@ -1,5 +1,7 @@
 package dev.ltms.bridged.inject;

+import dev.ltms.bridged.msg.TurnToken;
+
 /**
 * Notified when a worker's delegated turn is observed to complete — a confirmed
 * {@code WORKING → IDLE} transition after a delivery. This is the CB-106 completion signal the
@@ -57,7 +59,7 @@ public interface TurnListener {
     * resolve the send with the previous turn's stale answer. A default no-op keeps the interface
     * functional for callers that don't scrape.
     */
-    default void onDelivered(String target) {
+    default void onDelivered(String target, TurnToken token) {
    }

    /** No-op default for callers that only need delivery, not completion signalling. */
@@ -59,7 +59,7 @@ public final class LeadLauncher {
    /**
     * @param agents herdr agent control (start, list)
     * @param spaces workspace / tab control (ensure, create, label, list)
-     * @param cfg    the loaded config — {@code fleet.leaders}, {@code profiles} and the lead pins
+     * @param cfg    the loaded config — {@code fleet.leaders}, {@code profiles} and each lead's tab
     */
    public LeadLauncher(AgentControl agents, WorkspaceControl spaces, BridgedConfig cfg) {
        this.agents = agents;
@@ -102,8 +102,8 @@ public final class LeadLauncher {
                continue;
            }
            if (!lead.isCreatable()) {
-                // A lead with a `terminal:` pin and no `profile:` is recognise-only by design: the
-                // operator opens it by hand. Say so once rather than looking like a silent failure.
+                // A lead with a `tab:` but no `profile:` is recognise-only by design: the operator
+                // opens it by hand. Say so once rather than looking like a silent failure.
                log.info("lead '{}' is not live, and names no profile — it can be recognised but not "
                        + "launched. Add `profile:` under fleet.leaders.{} to have bridged start it.",
                        name, name);
@@ -127,18 +127,16 @@ public final class LeadLauncher {
    }

    /**
-     * How many live leads exist per configured name.
+     * 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.
     *
-     * <p>Two independent pieces of evidence, because either alone double-spawns:
-     * <ul>
-     *   <li>a running agent in a tab labelled {@code "<tabPrefix> <name>"} — how an auto-launched
-     *       lead, or an operator following the labelling convention, is found;</li>
-     *   <li>a running agent on a terminal the config pins in {@code fleet.leaders.<name>.terminal} —
-     *       how a lead the operator opened and pinned by hand is found. Without this, a pinned lead
-     *       whose tab carries no matching label would be relaunched on every boot.</li>
-     * </ul>
-     * 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.
     */
    private Map<String, Integer> liveLeads(Map<String, BridgedConfig.Leader> leaders) {
        Set<String> memberSpaces = cfg.profiles().values().stream()
@@ -160,20 +158,9 @@ public final class LeadLauncher {
            }
        }

-        // terminalId → the lead name the config pins it to.
-        Map<String, String> nameByPinnedTerminal = new LinkedHashMap<>();
-        leaders.forEach((name, lead) -> {
-            if (lead.terminal() != null && !lead.terminal().isBlank()) {
-                nameByPinnedTerminal.put(lead.terminal().strip(), name);
-            }
-        });
-
        Map<String, Integer> counts = new LinkedHashMap<>();
        for (Agent a : agents.list()) {
            String name = nameByTab.get(a.tabId());
-            if (name == null) {
-                name = nameByPinnedTerminal.get(a.terminalId());
-            }
            if (name != null) {
                counts.merge(name, 1, Integer::sum);
            }
@@ -184,7 +171,7 @@ public final class LeadLauncher {
    /**
     * The configured lead a tab label names, or {@code null} for a label that names none.
     *
-     * <p>Matched against the declared lead names rather than by splitting on the prefix, so an
+     * <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.
     */
    private String leadNameOf(String label, Map<String, BridgedConfig.Leader> leaders) {
@@ -193,7 +180,8 @@ public final class LeadLauncher {
        }
        String l = label.strip();
        for (Map.Entry<String, BridgedConfig.Leader> e : leaders.entrySet()) {
-            if (l.equalsIgnoreCase(e.getValue().tabLabel(e.getKey()).strip())) {
+            String tab = e.getValue().tabLabel();
+            if (tab != null && l.equalsIgnoreCase(tab.strip())) {
                return e.getKey();
            }
        }
@@ -202,7 +190,7 @@ public final class LeadLauncher {

    /** Start one lead. Returns false (having logged) rather than throwing on any failure. */
    private boolean launch(String name, BridgedConfig.Leader lead, BridgedConfig.Profile profile) {
-        String label = lead.tabLabel(name);
+        String label = lead.tabLabel();
        String cwd = (lead.cwd() == null || lead.cwd().isBlank())
                ? System.getProperty("user.dir") : lead.cwd();

@@ -458,6 +458,11 @@ public final class BridgeMcp {
                    "[worker finished without a structured bridge_reply — transcript tail follows]\n" + r.text());
            // The worker ran the turn then wedged (CB-109) — surface the error context.
            case WORKER_FAILED -> text("[worker failed — turn ended in an unrecoverable state]\n" + r.text());
+            // The backend refused on a subscription usage limit (CB-578 stage A) — the worker's
+            // pane stayed healthy, but its account is exhausted. Distinct from WORKER_FAILED so the
+            // primary gets the real cause, not a generic wedge.
+            case BACKEND_EXHAUSTED -> text("[backend exhausted — the worker's account refused on a "
+                    + "usage limit]\n" + r.text());
            // The worker paused mid-turn to ask (CB-205) — tell the primary how to answer in-turn.
            case QUESTION -> text("[question] the worker paused to ask before it can finish:\n" + r.text()
                    + "\n\nAnswer it by calling bridge_send again with turnId=\"" + r.turnId()
@@ -69,6 +69,14 @@ public final class MessageService {
         * failure context (e.g. the error screen). Terminal, but not a successful completion.
         */
        WORKER_FAILED,
+        /**
+         * The turn finished without a {@code bridge_reply} and the scrape matched the backend's
+         * configured usage-limit refusal pattern (CB-578 stage A); {@code text} is the reason,
+         * carrying the matched line. The worker's pane is healthy — only its account is refusing —
+         * so this is never reported as a completed reply, and is kept distinct from
+         * {@link #WORKER_FAILED} (a wedged worker) and a session simply going {@code GONE}.
+         */
+        BACKEND_EXHAUSTED,
        /**
         * The worker paused mid-turn to ask the primary a question (CB-205); {@code text} is the
         * question and {@code turnId} correlates the answer. Not terminal — the primary answers with
@@ -280,6 +288,7 @@ public final class MessageService {
            case COMPLETED_UNREPLIED -> "completion_fallback";
            case TIMED_OUT_WORKING, TIMED_OUT_QUEUED, BUSY -> "timeout";
            case WORKER_FAILED -> "failed";
+            case BACKEND_EXHAUSTED -> "backend_exhausted";
            case STALE_TURN, QUESTION -> null; // not a completed delegation
        };
    }
@@ -388,13 +397,14 @@ public final class MessageService {
                if (task != null) {
                    asyncTasksByWaiter.put(reply, task);
                }
+                TurnToken token = new TurnToken(target, reply);
                // The send has won the lock; the accepted-delivery hook records delegator ownership
                // here (CB-548). It runs BEFORE enqueue so a throwing hook — onAccepted is now a
                // public callback — fails the send without queuing a message that would orphan.
                if (onAccepted != null) {
                    onAccepted.run();
                }
-                CompletableFuture<Void> delivered = injector.enqueue(target, content);
+                CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
                try {
                    Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
                    return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
@@ -590,9 +600,11 @@ public final class MessageService {
            String source = r.outcome() == Outcome.REPLIED ? "reply" : "transcript";
            return new TaskView(ticket, Phase.DONE, r.text(), source, null, null);
        }
-        // A wedged worker (CB-109) carries the error context as its reason; the timeout/busy
-        // outcomes carry none, so fall back to the outcome name.
-        String detail = r.outcome() == Outcome.WORKER_FAILED && r.text() != null
+        // A wedged worker (CB-109) or a backend-exhausted classification (CB-578 stage A) carries
+        // the real cause as its reason; the timeout/busy outcomes carry none, so fall back to the
+        // outcome name.
+        boolean carriesReason = r.outcome() == Outcome.WORKER_FAILED || r.outcome() == Outcome.BACKEND_EXHAUSTED;
+        String detail = carriesReason && r.text() != null
                ? r.text()
                : "no reply — " + r.outcome().name().toLowerCase();
        return new TaskView(ticket, Phase.FAILED, null, null, detail, null);
@@ -668,6 +680,7 @@ public final class MessageService {
            case REPLY -> Outcome.REPLIED;
            case COMPLETION -> Outcome.COMPLETED_UNREPLIED;
            case FAILED -> Outcome.WORKER_FAILED;
+            case BACKEND_EXHAUSTED -> Outcome.BACKEND_EXHAUSTED;
            case QUESTION -> Outcome.QUESTION;
        };
    }
@@ -33,6 +33,13 @@ public final class Rendezvous {
        COMPLETION,
        /** The worker ran the turn then wedged (CB-109); {@code text} is the failure context. */
        FAILED,
+        /**
+         * The turn finished without a {@code bridge_reply}, and the scrape matched the backend's
+         * configured usage-limit refusal pattern (CB-578 stage A); {@code text} is the reason,
+         * carrying the matched line. The pane is healthy — only the account is refusing — so this
+         * is kept separate from a session simply going {@code GONE}.
+         */
+        BACKEND_EXHAUSTED,
        /**
         * The worker paused mid-turn to ask the primary a question (CB-205 reverse rendezvous);
         * {@code text} is the question and {@code turnId} correlates the primary's answer back to
@@ -224,6 +231,19 @@ public final class Rendezvous {
        return waiter != null && waiter.complete(new Resolution(Kind.FAILED, reason));
    }

+    /**
+     * Resolve a specific captured {@code waiter} as {@link Kind#BACKEND_EXHAUSTED} (CB-578 stage A):
+     * the turn finished with no {@code bridge_reply} and the scrape matched the backend's configured
+     * usage-limit pattern; {@code reason} carries the matched line. Like
+     * {@link #resolveCompletion(CompletableFuture, String)} it targets the exact captured send
+     * (CB-116). A no-op if that waiter was already resolved — first resolution wins.
+     *
+     * @return {@code true} if this call resolved the waiter, {@code false} if it was null or already resolved
+     */
+    public boolean resolveExhausted(CompletableFuture<Resolution> waiter, String reason) {
+        return waiter != null && waiter.complete(new Resolution(Kind.BACKEND_EXHAUSTED, reason));
+    }
+
    private boolean complete(String session, Resolution resolution) {
        CompletableFuture<Resolution> waiter = waiters.get(session);
        return waiter != null && waiter.complete(resolution);
@@ -0,0 +1,20 @@
+package dev.ltms.bridged.msg;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Identity for one accepted send. The session turn is deliberately absent: CompletionResolver's
+ * delivery callback runs before SessionManager.onDelivered, so binding it needs a later ordering design.
+ */
+public final class TurnToken {
+    private final String target;
+    private final CompletableFuture<Rendezvous.Resolution> waiter;
+
+    public TurnToken(String target, CompletableFuture<Rendezvous.Resolution> waiter) {
+        this.target = target;
+        this.waiter = waiter;
+    }
+
+    public String target() { return target; }
+    public CompletableFuture<Rendezvous.Resolution> waiter() { return waiter; }
+}
@@ -403,9 +403,12 @@ public final class BridgedApp {
                        case TIMED_OUT_QUEUED -> "queued";
                        case BUSY -> "busy";
                        case WORKER_FAILED -> "failed";
+                        case BACKEND_EXHAUSTED -> "backend_exhausted";
                        default -> "done"; // unreachable (terminal outcomes handled above)
                    },
-                    "detail", reply.outcome() == MessageService.Outcome.WORKER_FAILED && reply.text() != null
+                    "detail", (reply.outcome() == MessageService.Outcome.WORKER_FAILED
+                                    || reply.outcome() == MessageService.Outcome.BACKEND_EXHAUSTED)
+                                    && reply.text() != null
                            ? reply.text()
                            : "no reply within " + timeout + "ms; poll status or retry"));
        }
@@ -167,6 +167,22 @@ public final class GitWorktrees implements Worktrees {
        exec("git", "-C", repoRoot, "worktree", "remove", "--force", worktreePath);
    }

+    @Override
+    public boolean hasUncommitted(String worktreePath) {
+        // A worktree that is already gone holds no work to lose, and it must not break teardown:
+        // git -C <missing-dir> status exits non-zero and would throw where release() is mid-way
+        // through stopping a pane. Mirror remove()'s already-gone tolerance by treating it as clean.
+        Path p = Path.of(worktreePath);
+        if (!Files.exists(p)) {
+            log.debug("worktree {} already gone — nothing can be uncommitted", worktreePath);
+            return false;
+        }
+        // No --untracked-files=no: the exact shape of the work lost in CB-576 was a new file
+        // that was never added, so an untracked-only worktree is still dirty.
+        String out = exec("git", "-C", worktreePath, "status", "--porcelain");
+        return !out.isBlank();
+    }
+
    @Override
    public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
        if (overlay == null || overlay.isEmpty()) {
@@ -4,6 +4,7 @@ import dev.ltms.bridged.auth.MemberLifecycle;
 import dev.ltms.bridged.herdr.Agent;
 import dev.ltms.bridged.inject.TurnListener;
 import dev.ltms.bridged.inject.MemberPresence;
+import dev.ltms.bridged.msg.TurnToken;
 import dev.ltms.bridged.peer.MemberRole;
 import dev.ltms.bridged.peer.PeerHandle;
 import dev.ltms.bridged.peer.PeerLauncher;
@@ -200,6 +201,16 @@ public final class SessionManager implements TurnListener {
                    removed.paneId(), removed.terminalId(), removed.state(), cause);
            if (preserveWorktree && removed.worktree() != null) {
                logPreservedForShutdown(removed);
+            } else if (removed.worktree() != null && worktrees.hasUncommitted(removed.worktree())) {
+                // CB-576: a release that would otherwise remove the worktree finds it holding
+                // uncommitted work the bridge cannot see. A worker that ends a turn without
+                // committing (normally because it stopped to ask a question or refused the turn)
+                // has its only copy of that work in the worktree. Remove would --force-delete it,
+                // so preserve the directory and tell an operator where to find it.
+                preserveWorktree = true;
+                log.warn("release {} preserves dirty worktree {} for pane={} terminal={}: "
+                        + "the worktree holds uncommitted changes that --force remove would destroy",
+                        cause, removed.worktree(), removed.paneId(), removed.terminalId());
            }
            // CB-516: a send still waiting on this worker can never be answered now. Tell the
            // listener BEFORE the pane is torn down, so a blocked caller fails fast with a real
@@ -425,7 +436,7 @@ public final class SessionManager implements TurnListener {
     * can be re-delivered for multi-turn reuse until it is released.
     */
    @Override
-    public void onDelivered(String target) {
+    public void onDelivered(String target, TurnToken token) {
        MemberSession current = findByTerminal(target);
        if (current == null) return;
        if (current.state() != MemberSession.State.READY && current.state() != MemberSession.State.DONE) {
@@ -10,6 +10,18 @@ public interface Worktrees {
    /** git -C <repoRoot> worktree remove --force <path>. Idempotent (already-gone tolerated). */
    void remove(String repoRoot, String worktreePath);

+    /**
+     * True when the worktree holds uncommitted changes the bridge cannot see: tracked
+     * modifications, staged files, or untracked files. {@code git status --porcelain} is the
+     * test; an empty result means clean. Callers use this to decide whether removing the
+     * worktree would silently destroy a worker's only copy of its work.
+     *
+     * <p>An already-gone worktree is reported as clean (no throw), matching {@link #remove}'s
+     * idempotent contract: a path that does not exist holds no work to lose, and must not break
+     * a teardown that is mid-way through stopping the pane.
+     */
+    boolean hasUncommitted(String worktreePath);
+
    /** Copy each existing overlay path repoRoot→worktree; mark tracked ones --skip-worktree. */
    void overlayParity(String repoRoot, String worktreePath, List<String> overlay);

@@ -194,7 +194,7 @@ class BridgedConfigTest {
                fleet:
                  leaders:
                    opus:
-                      terminal: term_opus
+                      tab: "lead: opus"
                """);

        BridgedConfig.Leader lead = BridgedConfig.load(f).fleet().leaders().get("opus");
@@ -212,7 +212,7 @@ class BridgedConfigTest {
                fleet:
                  leaders:
                    opus:
-                      terminal: term_opus
+                      tab: "drive: opus"
                      tabPrefix: "drive:"
                      scanIntervalSeconds: 30
                """);
@@ -224,7 +224,9 @@ class BridgedConfigTest {

    /**
     * The pane no longer has to exist before the daemon does (CB-557): a lead naming a profile may
-     * be launched, while one that names only a terminal is recognised and never created.
+     * be launched, while one that names no profile is recognised and never created. Either way it
+     * still needs its own {@code tab:} (CB-579) — that part is unconditional, see
+     * {@link #aLeadWithNoTabRefusesToStart}.
     */
    @Test
    void aLeadIsCreatableOnlyWhenItNamesAProfile(@TempDir Path dir) throws Exception {
@@ -239,8 +241,9 @@ class BridgedConfigTest {
                  leaders:
                    launched:
                      profile: opus
+                      tab: "lead: launched"
                    pinned:
-                      terminal: term_opus
+                      tab: "lead: pinned"
                """);

        var leaders = BridgedConfig.load(f).fleet().leaders();
@@ -249,8 +252,12 @@ class BridgedConfigTest {
                "no profile to launch on ⇒ recognise-only, the pre-CB-557 behaviour");
    }

+    /**
+     * CB-579: {@code tab} is the only field a lead's identity depends on now, so it is required
+     * whether the entry is creatable or recognise-only — without it the entry can never be found.
+     */
    @Test
-    void aLeadThatCanBeNeitherFoundNorCreatedRefusesToStart(@TempDir Path dir) throws Exception {
+    void aLeadWithNoTabRefusesToStart(@TempDir Path dir) throws Exception {
        Path f = dir.resolve("useless-lead.yaml");
        Files.writeString(f, """
                bind:
@@ -264,6 +271,80 @@ class BridgedConfigTest {

        IllegalStateException e = assertThrows(IllegalStateException.class, cfg::validateMembers);
        assertTrue(e.getMessage().contains("ghost"), "the message must name the useless entry");
+        assertTrue(e.getMessage().contains("tab:"), "the message must say what is missing");
+    }
+
+    /**
+     * CB-579 acceptance (2): a config still spelling {@code fleet.leaders.<name>.terminal} must fail
+     * loudly at load, not be silently dropped by {@code Leader}'s {@code @JsonIgnoreProperties}.
+     */
+    @Test
+    void aLeaderTerminalKeyFailsLoadAndNamesTabAsTheReplacement(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("stale-terminal.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                fleet:
+                  leaders:
+                    opus:
+                      terminal: term_opus
+                """);
+
+        IllegalStateException e =
+                assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
+        assertTrue(e.getMessage().contains("opus"), "the message must name the offending entry");
+        assertTrue(e.getMessage().contains("tab:"), "the message must name the replacement key");
+        assertTrue(e.getMessage().contains("terminal"), "the message must name the retired key");
+    }
+
+    /** The same refusal, and it must name every offending entry, not just the first. */
+    @Test
+    void everyLeaderStillUsingTerminalIsReportedAtOnce(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("stale-terminals.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                fleet:
+                  leaders:
+                    opus:
+                      terminal: term_opus
+                    sol:
+                      terminal: term_sol
+                """);
+
+        IllegalStateException e =
+                assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
+        assertTrue(e.getMessage().contains("opus"));
+        assertTrue(e.getMessage().contains("sol"));
+    }
+
+    /** A {@code terminal:} anywhere else in the document (not under a leader entry) is unaffected. */
+    @Test
+    void aTerminalKeyOutsideFleetLeadersIsNotRejected(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("primary-terminal-ok.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\nprimary:\n  terminal: term_fixed\n");
+
+        assertDoesNotThrow(() -> BridgedConfig.load(f));
+    }
+
+    /** CB-579 acceptance (3): distinct `tab:` labels need no shared prefix — one scanner finds both. */
+    @Test
+    void twoLeadersWithDifferentTabsAreBothConfigured(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("two-tabs.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                fleet:
+                  leaders:
+                    opus:
+                      tab: "lead: opus"
+                    sol:
+                      tab: "captain: sol"
+                """);
+
+        var leaders = BridgedConfig.load(f).fleet().leaders();
+        assertEquals("lead: opus", leaders.get("opus").tab());
+        assertEquals("captain: sol", leaders.get("sol").tab());
    }

    // ── CB-551: the idle-lead heartbeat ─────────────────────────────────────────────────────────
@@ -328,7 +409,7 @@ class BridgedConfigTest {
                fleet:
                  leaders:
                    opus:
-                      terminal: term_opus
+                      tab: "lead: opus"
                      tabPrefix: "lead:"
                """);
        BridgedConfig cfg = BridgedConfig.load(f);
@@ -349,7 +430,7 @@ class BridgedConfigTest {
                  tabLabel: "lead: {role} {profile}"
                  leaders:
                    opus:
-                      terminal: term_opus
+                      tab: "lead: opus"
                """);
        BridgedConfig cfg = BridgedConfig.load(f);

@@ -374,7 +455,7 @@ class BridgedConfigTest {
                fleet:
                  leaders:
                    opus:
-                      terminal: term_opus
+                      tab: "lead: opus"
                """);

        assertDoesNotThrow(() -> BridgedConfig.load(f).validateLeadTabPrefixes());
@@ -400,7 +481,7 @@ class BridgedConfigTest {
                "a label that collides with a convention nobody reads is not a problem");
    }

-    // ── CB-530: the leaders registry ────────────────────────────────────────────────────────────
+    // ── CB-530/CB-579: the leaders registry ─────────────────────────────────────────────────────

    @Test
    void leadersBlockRegistersEveryPaneByName(@TempDir Path dir) throws Exception {
@@ -411,10 +492,10 @@ class BridgedConfigTest {
                fleet:
                  leaders:
                    opus-5.0:
-                      terminal: term_opus
+                      tab: "lead: opus-5.0"
                      kind: claude
                    gpt-sol-5.6:
-                      terminal: term_sol
+                      tab: "lead: gpt-sol-5.6"
                      kind: opencode
                      model: openai/gpt-5.6-terra
                """);
@@ -425,9 +506,9 @@ class BridgedConfigTest {
        assertEquals(Set.of("opus-5.0", "gpt-sol-5.6"), leaders.keySet());
        assertEquals("opencode", leaders.get("gpt-sol-5.6").kind());
        assertEquals("openai/gpt-5.6-terra", leaders.get("gpt-sol-5.6").model());
-        // The whole point: BOTH panes resolve as leads, so neither is demoted to worker.
-        assertEquals(Map.of("term_opus", "opus-5.0", "term_sol", "gpt-sol-5.6"),
-                cfg.leaderTerminals());
+        // Identity is the tab now (CB-579) — both entries carry their own, distinct label.
+        assertEquals("lead: opus-5.0", leaders.get("opus-5.0").tab());
+        assertEquals("lead: gpt-sol-5.6", leaders.get("gpt-sol-5.6").tab());
    }

    @Test
@@ -439,42 +520,6 @@ class BridgedConfigTest {
                "configs that never migrate must behave exactly as they did before CB-530");
    }

-    @Test
-    void anExplicitLeadersEntryWinsOverThePinForTheSameTerminal(@TempDir Path dir) throws Exception {
-        Path f = dir.resolve("both.yaml");
-        Files.writeString(f, """
-                bind:
-                  port: 8080
-                primary:
-                  terminal: term_shared
-                fleet:
-                  leaders:
-                    opus-5.0:
-                      terminal: term_shared
-                """);
-
-        assertEquals(Map.of("term_shared", "opus-5.0"), BridgedConfig.load(f).leaderTerminals(),
-                "the pin is the older spelling of the same fact; the named entry is what was meant");
-    }
-
-    @Test
-    void bothBlocksTogetherRegisterTheUnionOfTheirTerminals(@TempDir Path dir) throws Exception {
-        Path f = dir.resolve("union.yaml");
-        Files.writeString(f, """
-                bind:
-                  port: 8080
-                primary:
-                  terminal: term_pinned
-                fleet:
-                  leaders:
-                    gpt-sol-5.6:
-                      terminal: term_sol
-                """);
-
-        assertEquals(Map.of("term_pinned", "primary", "term_sol", "gpt-sol-5.6"),
-                BridgedConfig.load(f).leaderTerminals());
-    }
-
    @Test
    void neitherBlockLeavesNothingRegistered(@TempDir Path dir) throws Exception {
        Path f = dir.resolve("none.yaml");
@@ -483,22 +528,25 @@ class BridgedConfigTest {
        assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty());
    }

-    /** A lead entry with no terminal identifies nothing — it must not register a null key. */
+    /**
+     * CB-579: {@code fleet.leaders} no longer feeds {@code leaderTerminals()} at all — a lead's
+     * identity comes from the live tab scan, not a config-held terminal map. This method now exists
+     * only for the {@code primary.terminal} fallback.
+     */
    @Test
-    void aLeadWithoutATerminalIsNotRegistered(@TempDir Path dir) throws Exception {
-        Path f = dir.resolve("no-terminal.yaml");
+    void fleetLeadersNeverContributesToLeaderTerminals(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("leaders-only.yaml");
        Files.writeString(f, """
                bind:
                  port: 8080
                fleet:
                  leaders:
-                    sketch:
-                      kind: opencode
-                    real:
-                      terminal: term_real
+                    opus-5.0:
+                      tab: "lead: opus-5.0"
                """);

-        assertEquals(Map.of("term_real", "real"), BridgedConfig.load(f).leaderTerminals());
+        assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty(),
+                "no primary.terminal pin ⇒ nothing registered, even with fleet.leaders configured");
    }

    // ── CB-548: the architects registry ────────────────────────────────────────────────────────
@@ -1092,6 +1140,7 @@ class BridgedConfigTest {
                    gitHostEnv: GITEA_HOST
                    weight: 0.5
                    maxLoad: 2
+                    exhaustedPattern: "usage limit has been reached"
                placement: weighted
                lifecycle:
                  idleTtlSeconds: 300
@@ -1123,6 +1172,8 @@ class BridgedConfigTest {
        assertEquals("GITEA_HOST", w.gitHostEnv());
        assertEquals(0.5f, w.weight(), 0.0001f, "weight binds as a float");
        assertEquals(2, w.maxLoad(), "maxLoad binds as an integer");
+        assertTrue(w.hasExhaustedPattern(), "exhaustedPattern binds and enables the CB-578 stage A classification");
+        assertEquals("usage limit has been reached", w.exhaustedPattern());

        assertEquals("weighted", cfg.placement(), "placement binds at the top level");
        assertEquals(300, cfg.lifecycle().idleTtlSeconds());
@@ -20,8 +20,10 @@ import org.slf4j.LoggerFactory;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executors;
+import java.util.function.BiConsumer;

 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;

 class FleetHealthMonitorTest {
    @Test void oneTickUsesOneFleetListForAnyRosterSize() {
@@ -37,7 +39,8 @@ class FleetHealthMonitorTest {
        herdr.calls.clear();
        MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox());
        var scheduler = Executors.newSingleThreadScheduledExecutor();
-        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, sessions::roster, messages, scheduler, () -> 1, 60);
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, sessions::roster, messages, scheduler, () -> 1, 60,
+                (_, _) -> { });
        monitor.tick();
        monitor.stop();
        assertEquals(1, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
@@ -49,7 +52,7 @@ class FleetHealthMonitorTest {
        var scheduler = Executors.newSingleThreadScheduledExecutor();
        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
                new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
-                scheduler, () -> 1, 60);
+                scheduler, () -> 1, 60, (_, _) -> { });
        monitor.tick();
        herdr.healthy(true);
        monitor.tick();
@@ -68,7 +71,7 @@ class FleetHealthMonitorTest {
            var scheduler = Executors.newSingleThreadScheduledExecutor();
            FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
                    new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
-                    scheduler, () -> 1, 60);
+                    scheduler, () -> 1, 60, (_, _) -> { });
            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
            monitor.stop();
@@ -78,4 +81,90 @@ class FleetHealthMonitorTest {
            logger.detachAppender(appender);
        }
    }
+
+    // --- CB-580: a member that reaches GONE/NEVER_READY must fail its waiting tickets
+
+    private static FleetHealthMonitor monitorWith(BiConsumer<String, String> failTarget) {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        var scheduler = Executors.newSingleThreadScheduledExecutor();
+        return new FleetHealthMonitor(agents, java.util.List::of,
+                new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
+                scheduler, () -> 1, 60, failTarget);
+    }
+
+    @Test void terminalTransitionFailsTheTargetOnce() {
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.stop();
+        assertEquals(1, failTarget.calls.size());
+        assertEquals("term_a", failTarget.calls.get(0).target());
+        assertTrue(failTarget.calls.get(0).reason().contains("GONE"));
+    }
+
+    @Test void neverReadyNamesItselfAsTheReason() {
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.NEVER_READY);
+        monitor.stop();
+        assertEquals(1, failTarget.calls.size());
+        assertTrue(failTarget.calls.get(0).reason().contains("NEVER_READY"));
+    }
+
+    @Test void stayingInATerminalStateProducesOneFailureNotN() {
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.stop();
+        assertEquals(1, failTarget.calls.size());
+    }
+
+    @Test void aNonTerminalFaultStateDoesNotFailTheTarget() {
+        RecordingFailTarget failTarget = new RecordingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
+        monitor.stop();
+        assertEquals(0, failTarget.calls.size());
+    }
+
+    @Test void failTargetRetryIsBounded() {
+        AlwaysThrowingFailTarget failTarget = new AlwaysThrowingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.stop();
+        assertEquals(FleetHealthMonitor.MAX_FAIL_TARGET_ATTEMPTS, failTarget.calls);
+    }
+
+    @Test void exhaustedRetryStillDoesNotRefireOnAnUnchangedTick() {
+        AlwaysThrowingFailTarget failTarget = new AlwaysThrowingFailTarget();
+        FleetHealthMonitor monitor = monitorWith(failTarget);
+        monitor.reportTransition("term_a", HealthState.GONE);
+        int afterFirstTransition = failTarget.calls;
+        monitor.reportTransition("term_a", HealthState.GONE);
+        monitor.stop();
+        assertEquals(afterFirstTransition, failTarget.calls);
+    }
+
+    private record RecordedCall(String target, String reason) { }
+
+    private static final class RecordingFailTarget implements BiConsumer<String, String> {
+        final java.util.List<RecordedCall> calls = new java.util.ArrayList<>();
+
+        @Override public void accept(String target, String reason) {
+            calls.add(new RecordedCall(target, reason));
+        }
+    }
+
+    private static final class AlwaysThrowingFailTarget implements BiConsumer<String, String> {
+        int calls = 0;
+
+        @Override public void accept(String target, String reason) {
+            calls++;
+            throw new RuntimeException("boom");
+        }
+    }
 }
@@ -15,8 +15,9 @@ import java.util.concurrent.atomic.AtomicLong;
 import static org.junit.jupiter.api.Assertions.*;

 /**
- * CB-531. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that an
- * operator-labelled tab is what makes a pane a lead, and — just as importantly — what does not.
+ * CB-531/CB-579. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that
+ * an operator-labelled tab matching a configured {@code tab:} is what makes a pane a lead, and —
+ * just as importantly — what does not, and that a stale entry does not linger forever.
 */
 class LeadTabScannerTest {

@@ -120,23 +121,48 @@ class LeadTabScannerTest {
                .pane("w9:p1", "w9:t1", "term_worker");
    }

-    private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> configured,
+    /** The {@code tab:} → name map {@code twoLeads()}'s two lead tabs are configured under. */
+    private static Map<String, String> twoLeadsConfigured() {
+        return Map.of("lead: opus-5.0", "opus-5.0", "lead: gpt-sol-5.6", "gpt-sol-5.6");
+    }
+
+    private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> tabToName,
                                   AtomicLong clock) {
-        return new LeadTabScanner(herdr, "lead:", Set.of("bridged-workers"), configured, TTL,
-                clock::get);
+        return new LeadTabScanner(herdr, tabToName, Set.of("bridged-workers"), TTL, clock::get);
    }

    @Test
-    void everyLabelledTabBecomesALeadNamedByItsLabel() {
-        Map<String, String> leads = scanner(twoLeads(), Map.of(), new AtomicLong()).get();
+    void everyConfiguredTabBecomesALeadNamedByItsEntry() {
+        Map<String, String> leads = scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong()).get();

        assertEquals(Map.of("term_opus", "opus-5.0", "term_gpt", "gpt-sol-5.6"), leads,
-                "two leads discovered from labels alone — no terminal_id was ever configured");
+                "two leads discovered by their configured tab — no terminal_id was ever configured");
    }

    @Test
-    void anUnlabelledTabContributesNothing() {
-        assertFalse(scanner(twoLeads(), Map.of(), new AtomicLong()).get().containsKey("term_notes"));
+    void anUnconfiguredTabContributesNothing() {
+        assertFalse(scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong())
+                .get().containsKey("term_notes"));
+    }
+
+    /**
+     * CB-579: matching is exact against the configured map now, not a shared prefix — two leads with
+     * completely different labels are both discovered by one scanner, no convention required.
+     */
+    @Test
+    void twoLeadsWithCompletelyDifferentLabelsAreBothDiscovered() {
+        TopologyHerdr herdr = new TopologyHerdr()
+                .workspace("w1", "main")
+                .tab("w1:t1", "w1", "orchestrator: opus")
+                .tab("w1:t2", "w1", "captain: sol")
+                .pane("w1:p1", "w1:t1", "term_opus")
+                .pane("w1:p2", "w1:t2", "term_sol");
+        Map<String, String> tabToName = Map.of("orchestrator: opus", "opus", "captain: sol", "sol");
+
+        Map<String, String> leads = scanner(herdr, tabToName, new AtomicLong()).get();
+
+        assertEquals(Map.of("term_opus", "opus", "term_sol", "sol"), leads,
+                "no shared prefix needed — each lead is matched by its own configured tab");
    }

    /**
@@ -148,25 +174,28 @@ class LeadTabScannerTest {
    void aTabInAWorkerSpaceIsNeverALeadEvenWhenItsLabelMatches() {
        TopologyHerdr herdr = twoLeads().tab("w9:t2", "w9", "lead: impostor")
                .pane("w9:p2", "w9:t2", "term_impostor");
+        Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
+        tabToName.put("lead: impostor", "impostor");

-        assertFalse(scanner(herdr, Map.of(), new AtomicLong()).get().containsKey("term_impostor"));
+        assertFalse(scanner(herdr, tabToName, new AtomicLong()).get().containsKey("term_impostor"));
    }

    @Test
-    void aBarePrefixNamesNobodyAndIsRejected() {
+    void aLabelWithNoConfiguredEntryIsIgnored() {
        TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
-                .tab("w1:t1", "w1", "lead:").pane("w1:p1", "w1:t1", "term_a");
+                .tab("w1:t1", "w1", "lead: nobody-configured").pane("w1:p1", "w1:t1", "term_a");

-        assertEquals(Map.of(), scanner(herdr, Map.of(), new AtomicLong()).get(),
-                "a lead with no name would resolve as PRIMARY with nothing to attribute it to");
+        assertEquals(Map.of(), scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get(),
+                "a label that names no configured lead resolves nobody");
    }

    @Test
-    void thePrefixMatchesCaseInsensitivelyAndTheNameIsTrimmed() {
+    void matchingIsCaseInsensitiveAndToleratesSurroundingWhitespace() {
        TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
-                .tab("w1:t1", "w1", "  LEAD:   opus-5.0  ").pane("w1:p1", "w1:t1", "term_a");
+                .tab("w1:t1", "w1", "  LEAD: Opus-5.0  ").pane("w1:p1", "w1:t1", "term_a");

-        assertEquals(Map.of("term_a", "opus-5.0"), scanner(herdr, Map.of(), new AtomicLong()).get());
+        assertEquals(Map.of("term_a", "opus-5.0"),
+                scanner(herdr, Map.of("lead: Opus-5.0", "opus-5.0"), new AtomicLong()).get());
    }

    @Test
@@ -175,18 +204,51 @@ class LeadTabScannerTest {
        // nothing bridged placed can land here (see the worker-space test above).
        TopologyHerdr herdr = twoLeads().pane("w1:p1b", "w1:t1", "term_opus_split");

-        assertEquals("opus-5.0", scanner(herdr, Map.of(), new AtomicLong()).get().get("term_opus_split"));
+        assertEquals("opus-5.0",
+                scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get().get("term_opus_split"));
    }

+    /**
+     * 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
+     * drops out on the very next scan.
+     */
    @Test
-    void anExplicitlyConfiguredLeadIsMergedInAndOutranksALabel() {
-        Map<String, String> configured = Map.of("term_opus", "pinned-name", "term_extra", "from-config");
+    void aTabNoLongerPresentDropsTheLeadOnTheNextScan() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
+        assertTrue(s.get().containsKey("term_opus"));

-        Map<String, String> leads = scanner(twoLeads(), configured, new AtomicLong()).get();
+        // The session behind term_opus restarted — herdr no longer reports that tab or pane at all.
+        herdr.tabs.remove("w1:t1");
+        herdr.panes.remove("w1:p1");
+        clock.addAndGet(TTL);

-        assertEquals("pinned-name", leads.get("term_opus"), "an explicit pin is the operator's last word");
-        assertEquals("from-config", leads.get("term_extra"), "a configured lead needs no tab at all");
-        assertEquals("gpt-sol-5.6", leads.get("term_gpt"));
+        assertFalse(s.get().containsKey("term_opus"),
+                "a stale entry must expire once the tab it named is gone, not be merged back forever");
+    }
+
+    /**
+     * CB-579 acceptance (5): the whole point of matching by tab instead of {@code terminal_id} — a
+     * restart changes the terminal, not the tab, so the lead resolves under the same name with no
+     * config edit.
+     */
+    @Test
+    void aLeadRestartingInTheSameTabResolvesUnderTheSameName() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
+        assertEquals("opus-5.0", s.get().get("term_opus"));
+
+        // The session restarts: herdr assigns the pane a new terminal_id, same tab (w1:t1).
+        herdr.panes.remove("w1:p1");
+        herdr.pane("w1:p1", "w1:t1", "term_opus_v2");
+        clock.addAndGet(TTL);
+
+        Map<String, String> leads = s.get();
+        assertEquals("opus-5.0", leads.get("term_opus_v2"), "the new terminal resolves immediately");
+        assertFalse(leads.containsKey("term_opus"), "the old terminal_id is simply gone, not carried");
    }

    // ── caching ─────────────────────────────────────────────────────────────────────────────────
@@ -195,7 +257,7 @@ class LeadTabScannerTest {
    void aSecondLookupWithinTheTtlDoesNotTouchHerdr() {
        TopologyHerdr herdr = twoLeads();
        AtomicLong clock = new AtomicLong();
-        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);

        s.get();
        int afterFirst = herdr.calls;
@@ -210,21 +272,23 @@ class LeadTabScannerTest {
    void aTabLabelledAfterStartupIsPickedUpOnceTheTtlExpires() {
        TopologyHerdr herdr = twoLeads();
        AtomicLong clock = new AtomicLong();
-        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
+        tabToName.put("lead: late-arrival", "late-arrival");
+        LeadTabScanner s = scanner(herdr, tabToName, clock);
        assertFalse(s.get().containsKey("term_notes"));

        herdr.tab("w1:t3", "w1", "lead: late-arrival"); // the operator renames their tab
        clock.addAndGet(TTL);

        assertEquals("late-arrival", s.get().get("term_notes"),
-                "the whole point over `leaders:`: no config edit, no restart");
+                "the whole point over a config-held terminal_id: no config edit, no restart");
    }

    @Test
    void aFailedScanKeepsTheLeadsAlreadyKnownRatherThanDemotingThem() {
        TopologyHerdr herdr = twoLeads();
        AtomicLong clock = new AtomicLong();
-        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
        Map<String, String> before = s.get();

        herdr.failing = true;
@@ -235,14 +299,14 @@ class LeadTabScannerTest {
    }

    @Test
-    void aFailedFirstScanStillHonoursTheConfiguredLeads() {
+    void aFailedFirstScanReturnsEmptyRatherThanThrowing() {
        TopologyHerdr herdr = twoLeads();
        herdr.failing = true;

-        Map<String, String> leads = scanner(herdr, Map.of("term_x", "opus-5.0"), new AtomicLong()).get();
+        Map<String, String> leads = scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get();

-        assertEquals(Map.of("term_x", "opus-5.0"), leads,
-                "config-named leads must not depend on herdr answering at all");
+        assertEquals(Map.of(), leads,
+                "with nothing scanned yet and no override to fall back on, the map is simply empty");
    }

    @Test
@@ -250,7 +314,7 @@ class LeadTabScannerTest {
        TopologyHerdr herdr = twoLeads();
        herdr.failing = true;
        AtomicLong clock = new AtomicLong();
-        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);

        s.get();
        int afterFirst = herdr.calls;
@@ -7,9 +7,14 @@ import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.msg.TestTurnTokens;
+import dev.ltms.bridged.msg.TurnToken;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;

+import java.util.Set;
+import java.util.regex.Pattern;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -21,7 +26,7 @@ class CompletionResolverTest {
    void skipsTheScrapeWhenNoSendIsWaiting() {
        FakeHerdr herdr = new FakeHerdr();
        Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        resolver.resolve("term_a", null); // no in-flight turn captured for this target

@@ -33,7 +38,7 @@ class CompletionResolverTest {
    void failSkipsTheScrapeWhenNoSendIsWaiting() {
        FakeHerdr herdr = new FakeHerdr();
        Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        resolver.fail("term_a", null); // no in-flight turn, and no registered waiter to fall back to

@@ -45,9 +50,9 @@ class CompletionResolverTest {
    void captureBaselineSkipsTheReadWhenNoSendIsWaiting() {
        FakeHerdr herdr = new FakeHerdr().readText("⏺ X\n❯ ");
        Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

-        resolver.captureBaseline("term_a"); // no send to attribute a later completion to
+        resolver.captureBaseline("term_a", TestTurnTokens.inert("term_a")); // no send to attribute a later completion to

        assertFalse(herdr.called("agent.read"),
                "with no waiting send there is no turn to baseline — skip the scrape");
@@ -134,7 +139,7 @@ class CompletionResolverTest {
        // send must NOT be resolved with the stale answer.
        FakeHerdr herdr = new FakeHerdr().readText("⏺ 391\n❯ ");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");   // a send is blocked on this turn
        // The turn as captured at delivery: its waiter, and the previous turn's answer still on screen.
@@ -149,7 +154,7 @@ class CompletionResolverTest {
    void resolvesACompletionWhoseScrapeChangedSinceDelivery() {
        FakeHerdr herdr = new FakeHerdr().readText("⏺ No, 391 = 17 × 23.\n❯ "); // the worker's real answer
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        // Delivery baseline was the previous turn's "391"; the scrape now differs → resolve.
@@ -166,7 +171,7 @@ class CompletionResolverTest {
        String block = "⏺ " + "x".repeat(CompletionResolver.MAX_SCRAPE_CHARS + 1) + "\n❯ ";
        FakeHerdr herdr = new FakeHerdr().readText(block);
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -180,7 +185,7 @@ class CompletionResolverTest {
    void leavesAnUnclippedCompletionPaneTailUnmarked() {
        FakeHerdr herdr = new FakeHerdr().readText("⏺ complete report\n❯ ");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -192,9 +197,9 @@ class CompletionResolverTest {
    void resolvesSynchronouslyBeforePostTurnContextClearing() {
        FakeHerdr herdr = new FakeHerdr().readText("⏺ previous answer\n❯ ");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
        var waiter = rendezvous.open("term_a");
-        resolver.captureBaseline("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
        herdr.readText("⏺ answer that /clear would erase\n❯ ");

        resolver.resolveBeforePostAction("term_a");
@@ -214,10 +219,10 @@ class CompletionResolverTest {
        String longBlock = "⏺ " + "x".repeat(CompletionResolver.MAX_SCRAPE_CHARS + 500) + "\n❯ ";
        FakeHerdr herdr = new FakeHerdr().readText(longBlock);
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");        // a send is blocked on this turn
-        resolver.captureBaseline("term_a");            // baseline is the clipped >cap block
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiter)); // baseline is the clipped >cap block
        var turn = resolver.inFlight("term_a");
        assertEquals(CompletionResolver.MAX_SCRAPE_CHARS, turn.baseline().length(),
                "the delivery baseline is clipped to the same cap resolve() applies to the tail");
@@ -234,7 +239,7 @@ class CompletionResolverTest {
        // No delivery baseline (e.g. the pre-turn read failed) ⇒ never suppress; the completion resolves.
        FakeHerdr herdr = new FakeHerdr().readText("⏺ hello\n❯ ");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -252,7 +257,7 @@ class CompletionResolverTest {
        // byte-identical guard would wrongly match the empty tail and suppress.
        FakeHerdr herdr = new FakeHerdr().healthy(false); // agent.read throws HerdrException
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        var turn = new CompletionResolver.InFlight(waiter, ""); // empty pane baselined at delivery
@@ -272,7 +277,7 @@ class CompletionResolverTest {
        // fail must not overwrite that value, and must not even scrape the worker — nobody needs it.
        FakeHerdr herdr = new FakeHerdr().readText("an error screen");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a");
        var turn = new CompletionResolver.InFlight(waiter, null);
@@ -293,7 +298,7 @@ class CompletionResolverTest {
        // fail falls back to the waiter currently registered on the Rendezvous and fails it.
        FakeHerdr herdr = new FakeHerdr().readText("stuck on an error screen");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiter = rendezvous.open("term_a"); // send registered, but no captureBaseline ever ran
        resolver.fail("term_a", null);          // no in-flight turn → fall back to the registered waiter
@@ -319,7 +324,7 @@ class CompletionResolverTest {
        try {
            FakeHerdr herdr = new FakeHerdr().readText("stuck on an error screen");
            Rendezvous rendezvous = new Rendezvous();
-            CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+            CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
            var waiter = rendezvous.open("term_a");

            resolver.fail("term_a", null);
@@ -347,7 +352,7 @@ class CompletionResolverTest {
        // scrape to turn N+1; targeting turn N's captured waiter makes the late completion a no-op.
        FakeHerdr herdr = new FakeHerdr().readText("⏺ turn N answer\n❯ ");
        Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());

        var waiterN = rendezvous.open("term_a"); // turn N's send
        // The turn as the injector captured it at delivery (waiter + pre-turn baseline).
@@ -368,4 +373,88 @@ class CompletionResolverTest {
                "turn N stays resolved by its own reply");
        assertTrue(rendezvous.isWaiting("term_a"), "turn N+1 is still awaiting its own resolution");
    }
+
+    // --- CB-578 stage A: backend-exhausted classification ---------------------------------
+
+    @Test
+    void classifiesAMatchingScrapeAsBackendExhaustedInsteadOfACompletedReply() {
+        String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertTrue(waiter.isDone(), "a matching scrape still resolves the blocked send");
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
+                "not reported as a completed reply — the classification is distinct");
+    }
+
+    @Test
+    void theExhaustedReasonCarriesTheMatchedLine() {
+        String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals("backend exhausted (usage limit): The usage limit has been reached. Try again later.",
+                waiter.getNow(null).text(), "the reason names the real cause and carries the matched line");
+    }
+
+    @Test
+    void aNonMatchingScrapeResolvesAsAnOrdinaryCompletion() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ complete report\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
+                "a scrape that does not match the pattern is an ordinary completion");
+        assertEquals("complete report", waiter.getNow(null).text());
+    }
+
+    @Test
+    void aProfileWithNoConfiguredPatternKeepsTodaysCompletionFallbackUnchanged() {
+        // Even a scrape that WOULD have matched some other profile's pattern must resolve as a
+        // plain completion when this target's own profile has none configured (CB-578 criterion 4).
+        String block = "⏺ The usage limit has been reached.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver =
+                new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
+                "no pattern configured for this target's profile ⇒ unchanged completion-fallback behaviour");
+        assertEquals("The usage limit has been reached.", waiter.getNow(null).text());
+    }
+
+    @Test
+    void coverageIsOffWhenNoProfileHasAPatternConfigured() {
+        assertEquals("off (no profile has an exhaustedPattern configured; profiles: [terra])",
+                CompletionResolver.coverage(Set.of("terra"), Set.of()));
+    }
+
+    @Test
+    void coverageIsFullWhenEveryProfileHasAPatternConfigured() {
+        assertEquals("full (all profiles configured: [gx10, terra])",
+                CompletionResolver.coverage(Set.of("terra", "gx10"), Set.of("terra", "gx10")));
+    }
+
+    @Test
+    void coverageIsPartialAndNamesWhichProfilesAreConfigured() {
+        assertEquals("partial (configured: [terra]; not configured: [gx10])",
+                CompletionResolver.coverage(Set.of("terra", "gx10"), Set.of("terra")));
+    }
 }
@@ -8,6 +8,7 @@ import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.HerdrException;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;

@@ -47,7 +48,7 @@ class InjectorTest {

    @Test
    void deliversWhenIdle() {
-        CompletableFuture<Void> f = injector.enqueue(T, "hello");
+        CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
        assertFalse(f.isDone(), "not delivered until an injectable status arrives");
        injector.onStatus(T, AgentStatus.IDLE);
        assertTrue(f.isDone());
@@ -59,7 +60,7 @@ class InjectorTest {
        // CB-113: idle alone is not enough — hold until the worker's MCP is connected (ready).
        java.util.Set<String> ready = new java.util.HashSet<>();
        Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE); // idle but not yet available → held out of the boot window
        assertEquals(List.of(), sent(), "must not deliver into a not-yet-available worker");
@@ -79,7 +80,7 @@ class InjectorTest {
    void resubmitsEnterWhenADeliveredMessageIsNotPickedUp() {
        // CB-113: the Enter at delivery can race the paste; while the worker stays idle (not picked
        // up), the injector re-nudges Enter so the pending paste submits.
-        injector.enqueue(T, "task");
+        injector.enqueue(T, "task", TestTurnTokens.inert(T));
        injector.onStatus(T, AgentStatus.IDLE); // deliver: paste + one Enter
        long afterDeliver = enterKeystrokes();

@@ -95,7 +96,7 @@ class InjectorTest {

    @Test
    void holdsWhileWorkingThenDeliversOnIdle() {
-        injector.enqueue(T, "later");
+        injector.enqueue(T, "later", TestTurnTokens.inert(T));
        injector.onStatus(T, AgentStatus.WORKING);
        assertEquals(List.of(), sent(), "must not inject mid-turn");
        injector.onStatus(T, AgentStatus.IDLE);
@@ -104,7 +105,7 @@ class InjectorTest {

    @Test
    void blockedIsInjectableButUnknownIsNot() {
-        injector.enqueue(T, "answer");
+        injector.enqueue(T, "answer", TestTurnTokens.inert(T));
        injector.onStatus(T, AgentStatus.UNKNOWN);
        assertEquals(List.of(), sent(), "unknown status is not safe to inject");
        injector.onStatus(T, AgentStatus.BLOCKED);
@@ -113,8 +114,8 @@ class InjectorTest {

    @Test
    void twoRapidDeliveriesNeverInterleave() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));

        // First idle window delivers only m1, even if idle is observed twice before pickup.
        injector.onStatus(T, AgentStatus.IDLE);
@@ -129,8 +130,8 @@ class InjectorTest {

    @Test
    void transientUnknownDoesNotReleaseThePickupLatch() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));
        injector.onStatus(T, AgentStatus.IDLE); // m1 sent, awaiting pickup
        assertEquals(List.of("m1"), sent());

@@ -145,8 +146,8 @@ class InjectorTest {

    @Test
    void missedPickupEdgeIsReleasedByGraceSoTheQueueNeverWedges() {
-        injector.enqueue(T, "m1");
-        injector.enqueue(T, "m2");
+        injector.enqueue(T, "m1", TestTurnTokens.inert(T));
+        injector.enqueue(T, "m2", TestTurnTokens.inert(T));
        injector.onStatus(T, AgentStatus.IDLE); // m1 sent
        assertEquals(List.of("m1"), sent());

@@ -158,9 +159,9 @@ class InjectorTest {

    @Test
    void fifoOrderAcrossManyTurns() {
-        injector.enqueue(T, "a");
-        injector.enqueue(T, "b");
-        injector.enqueue(T, "c");
+        injector.enqueue(T, "a", TestTurnTokens.inert(T));
+        injector.enqueue(T, "b", TestTurnTokens.inert(T));
+        injector.enqueue(T, "c", TestTurnTokens.inert(T));
        for (int i = 0; i < 3; i++) {
            injector.onStatus(T, AgentStatus.IDLE);   // deliver one
            injector.onStatus(T, AgentStatus.WORKING); // pickup
@@ -172,7 +173,7 @@ class InjectorTest {
    @Test
    void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
        assertTrue(injector.activeTargets().isEmpty());
-        injector.enqueue(T, "x");
+        injector.enqueue(T, "x", TestTurnTokens.inert(T));
        assertEquals(Set.of(T), injector.activeTargets(), "active while a message is queued");

        injector.onStatus(T, AgentStatus.IDLE); // delivers; awaiting pickup
@@ -191,7 +192,7 @@ class InjectorTest {
    void firesTurnCompleteOnAConfirmedWorkingThenIdle() {
        List<String> completed = new ArrayList<>();
        Injector inj = new Injector(new AgentControl(herdr), completed::add);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);    // deliver
        inj.onStatus(T, AgentStatus.WORKING); // pickup + turn running
@@ -226,8 +227,8 @@ class InjectorTest {
        }
        ResetListener listener = new ResetListener();
        Injector inj = new Injector(agents, listener);
-        inj.enqueue(T, "first");
-        inj.enqueue(T, "second");
+        inj.enqueue(T, "first", TestTurnTokens.inert(T));
+        inj.enqueue(T, "second", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);    // first delegation
        inj.onStatus(T, AgentStatus.WORKING);
@@ -246,7 +247,7 @@ class InjectorTest {
    void doesNotSynthesizeCompletionFromAnUnconfirmedTurn() {
        List<String> completed = new ArrayList<>();
        Injector inj = new Injector(new AgentControl(herdr), completed::add);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        // Deliver, then only ever idle — a `working` sample is never seen. The pickup grace unwedges
        // the queue but must NOT invent a completion: without a sampled turn there is no trustworthy
@@ -285,7 +286,7 @@ class InjectorTest {
    void failsAnOutstandingDelegationWhoseWorkerWedgesInUnknown() {
        Captor cap = new Captor();
        Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);    // deliver
        inj.onStatus(T, AgentStatus.WORKING); // worker starts the turn
@@ -300,7 +301,7 @@ class InjectorTest {
    void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
        Captor cap = new Captor();
        Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);    // deliver
        inj.onStatus(T, AgentStatus.WORKING); // confirmed turn
@@ -315,7 +316,7 @@ class InjectorTest {
    void sendFailureDropsMessageAndFailsItsFuture() {
        FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
        Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom");
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);
        assertTrue(f.isCompletedExceptionally());
@@ -324,7 +325,7 @@ class InjectorTest {

    @Test
    void dropFailsPendingWaiters() {
-        CompletableFuture<Void> f = injector.enqueue(T, "orphan");
+        CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
        injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
        assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
    }
@@ -333,8 +334,8 @@ class InjectorTest {
    void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
        Captor cap = new Captor();
        Injector inj = new Injector(new AgentControl(herdr), cap);
-        CompletableFuture<Void> delivered = inj.enqueue(T, "delivered");
-        CompletableFuture<Void> queued = inj.enqueue(T, "queued");
+        CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
+        CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));

        inj.onStatus(T, AgentStatus.IDLE);    // deliver the first message
        inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
@@ -352,7 +353,7 @@ class InjectorTest {
        // leave its send hanging. A vanished worker must fail that in-flight turn too.
        Captor cap = new Captor();
        Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
        inj.onStatus(T, AgentStatus.IDLE);    // deliver
        inj.onStatus(T, AgentStatus.WORKING); // turn running

@@ -367,7 +368,7 @@ class InjectorTest {
        // by an off-sub worker's review of CB-110, delegated through the bridge.)
        Captor cap = new Captor();
        Injector inj = new Injector(new AgentControl(herdr), cap);
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));
        inj.onStatus(T, AgentStatus.IDLE); // deliver; pickup never confirmed

        inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
@@ -386,7 +387,7 @@ class InjectorTest {
        Captor cap = new Captor();
        List<String> forgotten = new ArrayList<>();
        Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
-        CompletableFuture<Void> f = inj.enqueue(T, "task");
+        CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));

        for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);

@@ -414,7 +415,7 @@ class InjectorTest {
        try {
            Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> false, _ -> {
            });
-            inj.enqueue(T, "task");
+            inj.enqueue(T, "task", TestTurnTokens.inert(T));

            for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);

@@ -439,7 +440,7 @@ class InjectorTest {
        Set<String> ready = new java.util.HashSet<>();
        Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains, _ -> {
        });
-        inj.enqueue(T, "task");
+        inj.enqueue(T, "task", TestTurnTokens.inert(T));

        for (int i = 0; i < 100; i++) inj.onStatus(T, AgentStatus.IDLE); // still booting, well under grace
        assertEquals(List.of(), sent());
@@ -455,7 +456,7 @@ class InjectorTest {
        // linger past the worker's life (MemberPresence.forget had no caller before this).
        List<String> forgotten = new ArrayList<>();
        Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, forgotten::add);
-        inj.enqueue(T, "orphan");
+        inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
        inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
        assertEquals(List.of(T), forgotten, "drop clears the gone worker's presence");
    }
@@ -476,7 +477,7 @@ class InjectorTest {
        try {
            Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, _ -> {
            });
-            inj.enqueue(T, "orphan");
+            inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
            inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));

            String warn = appender.list.stream()
@@ -500,7 +501,7 @@ class InjectorTest {
        StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
        poller.start();
        try {
-            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller");
+            CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
            delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
        } finally {
            poller.stop();
@@ -519,7 +520,7 @@ class InjectorTest {
    void deliveredFutureCarriesSendFailure() {
        FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
        Injector inj = new Injector(new AgentControl(failing));
-        CompletableFuture<Void> f = inj.enqueue(T, "boom");
+        CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
        inj.onStatus(T, AgentStatus.IDLE);
        ExecutionException ex = assertThrows(ExecutionException.class, f::get);
        assertInstanceOf(HerdrException.class, ex.getCause());
@@ -28,7 +28,7 @@ class LeadLauncherTest {
                List.of("ccs", "ltms"), "tab", "bridged-workers", null,
                "http://127.0.0.1:8765/mcp", null, null,
                null, null, null,
-                Map.of("CLAUDE_CODE_AUTO_COMPACT_WINDOW", "300000"), null, null, true);
+                Map.of("CLAUDE_CODE_AUTO_COMPACT_WINDOW", "300000"), null, null, true, null);
    }

    private static BridgedConfig configWith(BridgedConfig.Leader lead) {
@@ -41,8 +41,8 @@ class LeadLauncherTest {
                null, null, fleet, null, "fixed", null).withDefaults();
    }

-    private static BridgedConfig.Leader lead(String profile, String terminal, int instances) {
-        return new BridgedConfig.Leader(profile, terminal, instances, "lead:", 10, null, null,
+    private static BridgedConfig.Leader lead(String profile, String tab, int instances) {
+        return new BridgedConfig.Leader(profile, tab, instances, "lead:", 10, null, null,
                "leads", "/repo");
    }

@@ -70,16 +70,16 @@ class LeadLauncherTest {
    void startsTheDeclaredLeadWhenNoneIsRunning() {
        FakeHerdr herdr = new FakeHerdr();

-        assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
        assertTrue(herdr.called("agent.start"), "a lead must actually be started");
        assertEquals("lead-opus", startedName(herdr));
    }

-    /** The tab is labelled so the scanner finds the lead on the next resolve. */
+    /** The tab is labelled with the configured `tab:` so the scanner finds the lead on the next resolve. */
    @Test
-    void labelsTheTabWithThePrefixTheScannerReadsBack() {
+    void labelsTheTabWithTheConfiguredTabValue() {
        FakeHerdr herdr = new FakeHerdr();
-        launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
+        launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();

        assertEquals("lead: opus",
                ((Map<?, ?>) herdr.lastCall("tab.rename").params()).get("label"));
@@ -90,7 +90,7 @@ class LeadLauncherTest {
    void startsAsManyInstancesAsAreDeclared() {
        FakeHerdr herdr = new FakeHerdr();

-        assertEquals(2, launcher(herdr, configWith(lead("opus", null, 2))).ensureLeads());
+        assertEquals(2, launcher(herdr, configWith(lead("opus", "lead: opus", 2))).ensureLeads());
        assertEquals(2, herdr.calls.stream().filter(c -> c.method().equals("agent.start")).count());
    }

@@ -104,7 +104,7 @@ class LeadLauncherTest {
                .withTab("wL", "wL:t1", "lead: opus")
                .withAgent("lead-opus", "term_lead", "wL:p1", "wL:t1");

-        assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
        assertFalse(herdr.called("agent.start"), "the live lead must not be duplicated");
    }

@@ -118,20 +118,23 @@ class LeadLauncherTest {
                .withWorkspace("wL", "leads")
                .withTab("wL", "wL:t1", "lead: opus"); // label only — nothing running in it

-        assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
                "a stale label is not a lead; the lead must be relaunched");
    }

    /**
-     * A lead the operator opened by hand and pinned with `terminal:` is live even though its tab
-     * carries no matching label. Counting labels alone would relaunch it on every boot.
+     * 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
+     * auto-launched one is, by its tab, not by a terminal id nobody wrote down in advance.
     */
    @Test
-    void aPinnedTerminalWithARunningAgentCountsAsLive() {
+    void aHandOpenedLeadWithTheConfiguredTabLabelCountsAsLive() {
        FakeHerdr herdr = new FakeHerdr()
-                .withAgent("hand-opened", "term_pinned", "wX:p1", "wX:t1");
+                .withWorkspace("wX", "main")
+                .withTab("wX", "wX:t1", "lead: opus")
+                .withAgent("hand-opened", "term_hand", "wX:p1", "wX:t1");

-        assertEquals(0, launcher(herdr, configWith(lead("opus", "term_pinned", 1))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
        assertFalse(herdr.called("agent.start"));
    }

@@ -143,7 +146,7 @@ class LeadLauncherTest {
                .withTab("wM", "wM:t1", "lead: opus")              // a member tab that looks like a lead
                .withAgent("claude-opus-x", "term_m", "wM:p1", "wM:t1");

-        assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
+        assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
                "a member in a lead-labelled tab is not a lead, so the real lead is still missing");
    }

@@ -152,7 +155,7 @@ class LeadLauncherTest {
    void anUncountableHerdrStartsNothing() {
        FakeHerdr herdr = new FakeHerdr().healthy(false);

-        assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
        assertFalse(herdr.called("agent.start"));
    }

@@ -166,7 +169,7 @@ class LeadLauncherTest {
    @Test
    void theLeadNeverReceivesTheWorkerReplyCharter() {
        FakeHerdr herdr = new FakeHerdr();
-        launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
+        launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();

        List<String> args = startedArgs(herdr);
        assertFalse(args.contains("--append-system-prompt"),
@@ -178,7 +181,7 @@ class LeadLauncherTest {
    @Test
    void theLeadMountsTheBridgeMcpAndPinsItsModel() {
        FakeHerdr herdr = new FakeHerdr();
-        launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
+        launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();

        List<String> args = startedArgs(herdr);
        assertTrue(args.contains("--mcp-config"));
@@ -192,7 +195,7 @@ class LeadLauncherTest {
    @Test
    void theLeadEnvCarriesNoAnthropicBinding() {
        FakeHerdr herdr = new FakeHerdr();
-        launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
+        launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();

        Map<String, String> env = tabEnv(herdr);
        assertNull(env.get("ANTHROPIC_BASE_URL"));
@@ -205,7 +208,7 @@ class LeadLauncherTest {
    @Test
    void theLeadTabIsCreatedOutsideEveryMemberWorkspace() {
        FakeHerdr herdr = new FakeHerdr();
-        launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
+        launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();

        String label = (String) ((Map<?, ?>) herdr.lastCall("workspace.create").params()).get("label");
        assertEquals("leads", label);
@@ -214,12 +217,12 @@ class LeadLauncherTest {

    // ── recognise-only and misconfiguration ───────────────────────────────────────────────────

-    /** A lead with a pin but no profile is recognise-only by design — not an error, not a launch. */
+    /** A lead with a tab but no profile is recognise-only by design — not an error, not a launch. */
    @Test
    void aLeadThatNamesNoProfileIsRecognisedButNeverLaunched() {
        FakeHerdr herdr = new FakeHerdr();

-        assertEquals(0, launcher(herdr, configWith(lead(null, "term_dead", 1))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead(null, "lead: dead", 1))).ensureLeads());
        assertFalse(herdr.called("agent.start"));
    }

@@ -228,7 +231,7 @@ class LeadLauncherTest {
    void zeroInstancesLaunchesNothing() {
        FakeHerdr herdr = new FakeHerdr();

-        assertEquals(0, launcher(herdr, configWith(lead("opus", null, 0))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 0))).ensureLeads());
        assertFalse(herdr.called("agent.start"));
    }

@@ -237,7 +240,7 @@ class LeadLauncherTest {
    void anUnknownProfileIsSkippedRatherThanThrown() {
        FakeHerdr herdr = new FakeHerdr();

-        assertEquals(0, launcher(herdr, configWith(lead("nope", null, 1))).ensureLeads());
+        assertEquals(0, launcher(herdr, configWith(lead("nope", "lead: opus", 1))).ensureLeads());
        assertFalse(herdr.called("agent.start"));
    }

@@ -733,7 +733,7 @@ class ClaudeCodeLauncherTest {
        return new BridgedConfig.Profile(
                profile, baseUrl, "sonnet", null, "BRIDGED_WORKER_TOKEN",
                List.of("ccs", profile), "tab", "bridged-workers", "w #{n}", null, null, null,
-                null, null, null, Map.of(), null, null, true);
+                null, null, null, Map.of(), null, null, true, null);
    }

    @Test
@@ -819,7 +819,7 @@ class ClaudeCodeLauncherTest {
                null, null, null,
                Map.of("ANTHROPIC_BASE_URL", "http://evil.example.com",
                        "ANTHROPIC_AUTH_TOKEN", "sk-ant-bad", "JAVA_HOME", "/opt/jdk"),
-                null, null, true);
+                null, null, true, null);
        new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
                _ -> "would-be-token").spawn();
@@ -5,6 +5,7 @@ import dev.ltms.bridged.herdr.AgentStatus;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.inject.CompletionResolver;
+import dev.ltms.bridged.inject.ExhaustedPatternLookup;
 import dev.ltms.bridged.mcp.PrimaryRegistry;
 import dev.ltms.bridged.inject.Injector;
 import org.junit.jupiter.api.BeforeEach;
@@ -31,7 +32,8 @@ class MessageServiceTest {
    private final FakeHerdr herdr = new FakeHerdr().readText("BUILD GREEN: 391 files");
    private final AgentControl agents = new AgentControl(herdr);
    private final Rendezvous rendezvous = new Rendezvous();
-    private final CompletionResolver completion = new CompletionResolver(agents, rendezvous);
+    private final CompletionResolver completion =
+            new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none());
    private final Injector injector = new Injector(agents, completion);
    private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
    private final MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
@@ -127,7 +129,7 @@ class MessageServiceTest {
        injector.onStatus(T, AgentStatus.IDLE);    // first delivery
        injector.onStatus(T, AgentStatus.WORKING); // first turn in flight

-        CompletableFuture<Void> queued = injector.enqueue(T, "second task");
+        CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
        injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));

@@ -114,6 +114,17 @@ class RendezvousTest {
                "the first resolution wins; the stored value is unchanged");
    }

+    @Test
+    void resolveExhaustedTwiceIsANoOpTheSecondTime() {
+        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.open(W);
+        assertTrue(rendezvous.resolveExhausted(waiter, "first reason"), "the first classification resolves");
+        assertFalse(rendezvous.resolveExhausted(waiter, "second reason"),
+                "a second exhausted resolution on an already-resolved waiter returns false");
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind());
+        assertEquals("first reason", waiter.getNow(null).text(),
+                "the first resolution wins; the stored value is unchanged");
+    }
+
    @Test
    void closeAskRemovesTheTurn() {
        Rendezvous.AskTicket t = rendezvous.openAsk(W);
@@ -0,0 +1,20 @@
+package dev.ltms.bridged.msg;
+
+/**
+ * Explicit unbound tokens for tests that exercise delivery without an accepted send.
+ *
+ * <p>The waiter is {@code null} on purpose. "No accepted send" is an <em>absence</em>, and a helper
+ * that handed back a fresh {@code CompletableFuture} would invent one — which is how the first
+ * version of this class turned {@code captureBaselineSkipsTheReadWhenNoSendIsWaiting} red: the
+ * resolver saw a non-null waiter, decided a turn was in flight, and scraped a pane that no send was
+ * blocked on. An inert value must omit the fact, never fabricate it.
+ */
+public final class TestTurnTokens {
+    private TestTurnTokens() {
+    }
+
+    /** A token for a delivery that no send is waiting on: it authorises nothing. */
+    public static TurnToken inert(String target) {
+        return new TurnToken(target, null);
+    }
+}
@@ -29,6 +29,7 @@ public final class FakeWorktrees implements Worktrees {
    private final Set<String> existingPaths = ConcurrentHashMap.newKeySet();
    private final Set<String> trackedPaths = ConcurrentHashMap.newKeySet();
    private volatile RuntimeException addFailure;
+    private volatile boolean dirty = false;
    private volatile String repoRoot = "/repo";
    private volatile String prefix = "/worktrees";

@@ -61,6 +62,12 @@ public final class FakeWorktrees implements Worktrees {
        return this;
    }

+    /** Mark the worktree dirty so {@link #hasUncommitted} reports true (simulates uncommitted work). */
+    public FakeWorktrees withDirty(boolean dirty) {
+        this.dirty = dirty;
+        return this;
+    }
+
    @Override
    public String add(String repoRoot, String branch, String baseRef) {
        addCalls.add(new AddCall(repoRoot, branch, baseRef));
@@ -77,6 +84,11 @@ public final class FakeWorktrees implements Worktrees {
        removeCalls.add(new RemoveCall(repoRoot, worktreePath));
    }

+    @Override
+    public boolean hasUncommitted(String worktreePath) {
+        return dirty;
+    }
+
    @Override
    public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
        List<String> copied = new java.util.ArrayList<>();
@@ -175,6 +175,46 @@ class GitWorktreesTest {
        assertTrue(Files.exists(Path.of(wt).resolve(".mcp.json")), ".mcp.json stub was dropped");
    }

+    /**
+     * CB-576. {@code hasUncommitted} must treat a freshly-provisioned worktree as clean, but a
+     * worktree holding a brand-new, never-added file as dirty. The untracked-file-only shape is
+     * exactly the work lost in the incident — a worker's draft that compiled but was never
+     * committed because it stopped to ask its lead a question.
+     */
+    @Test
+    void anUntrackedOnlyWorktreeCountsAsDirty(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
+        String wt = gitWorktrees.add(repo.toString(), "cb-576-u", "HEAD");
+
+        assertFalse(gitWorktrees.hasUncommitted(wt),
+                "a freshly provisioned worktree must read as clean");
+
+        Files.writeString(Path.of(wt).resolve("brand-new.txt"), "draft that was never added\n");
+
+        assertTrue(gitWorktrees.hasUncommitted(wt),
+                "an untracked-only file must count as dirty");
+
+        Files.writeString(Path.of(wt).resolve("README.md"), "edited tracked file\n");
+        assertTrue(gitWorktrees.hasUncommitted(wt),
+                "a tracked modification must also count as dirty");
+    }
+
+    /**
+     * CB-576 review. {@code hasUncommitted} must tolerate a missing worktree exactly like
+     * {@code remove}: an already-gone directory holds no work to lose, and throwing here would
+     * break teardown — SessionManager.release() calls it before stopping the pane, so an
+     * exception would orphan a live pane and skip the release notification (CB-516).
+     */
+    @Test
+    void hasUncommittedOnAMissingWorktreeReturnsFalseWithoutThrowing(@TempDir Path tmp) {
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
+        String gone = tmp.resolve("wts").resolve("does-not-exist").toString();
+
+        assertFalse(gitWorktrees.hasUncommitted(gone),
+                "a missing worktree is reported clean, not an error");
+    }
+
    /** All three protected configs are covered: each one present in a worktree is neutralized and hidden. */
    @Test
    void allThreeConfigsAreNeutralizedWhenPresent(@TempDir Path tmp) throws Exception {
@@ -10,6 +10,7 @@ import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.WorkspaceControl;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import dev.ltms.bridged.peer.PeerUnreachableException;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
@@ -101,7 +102,7 @@ class SessionManagerTest {

        assertDoesNotThrow(() -> sessions.asPresence().markPresent(null),
                "the primary's null terminal must not blow up an unrelated tool call");
-        assertDoesNotThrow(() -> sessions.onDelivered(null));
+        assertDoesNotThrow(() -> sessions.onDelivered(null, TestTurnTokens.inert(null)));
        assertDoesNotThrow(() -> sessions.onTurnComplete(null));
        assertDoesNotThrow(() -> sessions.onTurnFailed(null));

@@ -121,7 +122,7 @@ class SessionManagerTest {
                "MCP presence moves SPAWNING → READY");
        assertTrue(sessions.asPresence().isPresent(terminal), "presence is also recorded");

-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
                "delivery moves READY → BUSY");

@@ -153,7 +154,7 @@ class SessionManagerTest {
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        String terminal = session.terminalId();
        sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));

        sessions.onTurnFailed(terminal);

@@ -181,7 +182,7 @@ class SessionManagerTest {
            MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
            String terminal = session.terminalId();
            sessions.asPresence().markPresent(terminal);
-            sessions.onDelivered(terminal);
+            sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));

            sessions.onTurnFailed(terminal);

@@ -264,7 +265,7 @@ class SessionManagerTest {
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        String terminal = session.terminalId();
        sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));

        clock[0] = 100;
        assertEquals(0, sessions.reapIdle(10), "BUSY session past TTL is never reaped");
@@ -281,7 +282,7 @@ class SessionManagerTest {
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        String terminal = session.terminalId();
        sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        sessions.onTurnComplete(terminal);

        clock[0] = 21;
@@ -299,7 +300,7 @@ class SessionManagerTest {
        MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "owner2");
        sessions.asPresence().markPresent(ready.terminalId());
        sessions.asPresence().markPresent(busy.terminalId());
-        sessions.onDelivered(busy.terminalId());
+        sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));

        clock[0] = 50;
        assertEquals(1, sessions.reapIdle(30), "only READY past TTL is reaped");
@@ -317,9 +318,9 @@ class SessionManagerTest {
        String terminal = session.terminalId();
        sessions.asPresence().markPresent(terminal);

-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        sessions.onTurnComplete(terminal);
-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        sessions.onTurnComplete(terminal);

        MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -337,13 +338,13 @@ class SessionManagerTest {
        String terminal = session.terminalId();
        sessions.asPresence().markPresent(terminal);

-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        sessions.onTurnComplete(terminal);
        assertEquals(MemberSession.State.DONE,
                sessions.get(session.paneId()).orElseThrow().state(),
                "first turn completes without release");

-        sessions.onDelivered(terminal);
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
        sessions.onTurnComplete(terminal);

        assertTrue(sessions.get(session.paneId()).isEmpty(), "session released after cap reached");
@@ -359,7 +360,7 @@ class SessionManagerTest {
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        sessions.asPresence().markPresent(session.terminalId());

-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
        assertTrue(sessions.onTurnCompleteWithPostAction(session.terminalId()));

        MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -373,7 +374,7 @@ class SessionManagerTest {
        SessionManager sessions = sessionManager(herdr, () -> 0L, 1, true);
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        sessions.asPresence().markPresent(session.terminalId());
-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));

        assertFalse(sessions.hasPostTurnAction(session.terminalId()),
                "a session at its cap will be released, not reset for reuse");
@@ -388,7 +389,7 @@ class SessionManagerTest {
        SessionManager sessions = sessionManager(herdr, () -> 0L, 0, false);
        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
        sessions.asPresence().markPresent(session.terminalId());
-        sessions.onDelivered(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));

        sessions.onTurnComplete(session.terminalId());

@@ -406,7 +407,7 @@ class SessionManagerTest {
        MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "ownerB");
        sessions.asPresence().markPresent(ready.terminalId());
        sessions.asPresence().markPresent(busy.terminalId());
-        sessions.onDelivered(busy.terminalId());
+        sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));

        sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));

@@ -6,9 +6,15 @@ import dev.ltms.bridged.guard.SubscriptionGuard;
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.WorkspaceControl;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.msg.TestTurnTokens;
 import dev.ltms.bridged.peer.MemberRole;
 import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;

 import java.util.List;
 import java.util.Map;
@@ -178,6 +184,49 @@ class WorktreeSessionManagerTest {
        assertTrue(sessions.get(paneId).isEmpty(), "released session is no longer retrievable");
    }

+    /**
+     * CB-576. A normal {@code COMPLETED} release whose worktree holds uncommitted work must NOT
+     * remove it — {@code --force} would destroy the worker's only copy. The bridge cannot see
+     * uncommitted files, so the worktree is preserved and the release logged at WARN naming the
+     * path, the session, and the cause an operator needs to find the work.
+     */
+    @Test
+    void releasePreservesDirtyWorktreeAndLogsWarn() {
+        FakeHerdr herdr = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
+                .withDirty(true);
+        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-576", null));
+
+        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.setContext(ctx);
+        appender.start();
+        sessionLog.addAppender(appender);
+        sessionLog.setLevel(Level.WARN);
+        try {
+            sessions.release(s.paneId());
+
+            assertTrue(herdr.called("pane.close"), "release still tears the worker pane down");
+            assertTrue(worktrees.removeCalls().isEmpty(),
+                    "a dirty worktree is never removed — it holds the only copy of the work");
+            String warn = appender.list.stream()
+                    .filter(e -> e.getLevel().equals(Level.WARN))
+                    .map(ILoggingEvent::getFormattedMessage)
+                    .filter(m -> m.contains("dirty worktree"))
+                    .findFirst()
+                    .orElse("no dirty-release WARN logged");
+            assertTrue(warn.contains(s.worktree()), "the WARN names the worktree path: " + warn);
+            assertTrue(warn.contains(s.terminalId()), "the WARN names the session: " + warn);
+            assertTrue(warn.contains("COMPLETED"), "the WARN names the release cause: " + warn);
+        } finally {
+            sessionLog.detachAppender(appender);
+        }
+    }
+
    @Test
    void drainAllPreservesWorktreeOfIdleSession() {
        FakeHerdr herdr = new FakeHerdr();
@@ -204,7 +253,7 @@ class WorktreeSessionManagerTest {
                new WorktreeRequest("cb-544", null));
        String terminal = s.terminalId();
        sessions.asPresence().markPresent(terminal);
-        sessions.onDelivered(terminal); // BUSY, never completes → still BUSY when the timeout hits
+        sessions.onDelivered(terminal, TestTurnTokens.inert(terminal)); // BUSY, never completes → still BUSY when the timeout hits

        sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));

@@ -237,7 +237,7 @@ state never presents stop as the only action.
 | Release cause | Process action | Provisioned worktree |
 |---|---|---|
 | `SPAWN_ROLLBACK` before registration or delivery | Stop and clean up | Remove |
-| `COMPLETED` for `READY` or `DONE` without pending work, idle TTL, or successful context-cap completion | Stop | Remove under completed policy |
+| `COMPLETED` for `READY` or `DONE` without pending work, idle TTL, or successful context-cap completion | Stop | Remove only if clean; preserve a dirty worktree (CB-576) |
 | `NEVER_READY` | Stop | Preserve |
 | `GONE` | Best-effort stop | Preserve |
 | `TURN_FAILED` or lead abort while `BUSY` or `FAILED` | Stop | Preserve |
Author	SHA1	Message	Date
Dai Ha	541df87272	CB-578 stage A: classify a usage-limit refusal instead of a completed reply CI / build (pull_request) Failing after 59s Details CI / contract (pull_request) Successful in 1m10s Details A backend that refuses on a subscription usage limit leaves the pane healthy but the turn ends with no bridge_reply; the completion fallback used to scrape and hand that refusal back as if it were a real answer. CompletionResolver now matches the scrape against a per-profile exhaustedPattern (config, never a vendor string) and resolves the send as Rendezvous.Kind/Outcome.BACKEND_EXHAUSTED with a reason carrying the matched line, kept distinct from GONE/WORKER_FAILED. A profile with no pattern configured is unaffected. Coverage is logged at startup via CompletionResolver.coverage(...), naming which profiles have a pattern and which don't, following FleetHealthMonitor.coverage's pattern.	2026-08-15 09:55:06 +02:00
ltms	42f46dfe9a	Merge CB-579: resolve a lead by its tab name, drop the terminal-id pin CI / build (push) Successful in 58s Details CI / contract (push) Successful in 1m4s Details Verified by the lead: merged onto main (`9088d2b`) in a scratch worktree, mvn -f bridged/pom.xml clean install unpiped — MVN_EXIT=0, Tests run: 696, Failures: 0, BUILD SUCCESS. main alone measures 692, so this adds 4 net tests. Merges cleanly; Bridged.java auto-merged against CB-580. Reviewed by the lead reading the full production diff and the three test files. The member's own report was lost to the idle reaper before collection, so there was no author write-up. Closes the live bug: LeadTabScanner.scan() no longer merges the config pin over the scan result, and the cache no longer seeds from it, so a lead disappears once its tab is gone. The ghost this fixes had begun throwing agent_not_found from ReplyPushLoop.decide on every tick, against a dead terminal that still owned two live members. Beyond the brief, and correct: `tab` is required for every leader, not only non-creatable ones, because LeadLauncher also uses it to label a tab it creates. terminal: is rejected by a raw-YAML check rather than by record shape — the only way to beat @JsonIgnoreProperties(ignoreUnknown = true). The silent-default trap is avoided: the back-compat constructor still takes `tab` positionally. Behaviour change worth knowing: the scanner's initial cache is now empty instead of the config pins, so a herdr failure on the very first scan yields no leads until a scan succeeds. That is unavoidable once the pins are gone, it fails loudly rather than silently, and it is covered by aFailedFirstScanReturnsEmptyRatherThanThrowing. Operator action required: bridged.yaml must replace fleet.leaders.<name>.terminal with tab. Already done for this deployment.	2026-08-15 09:29:00 +02:00
ltms	9088d2b2c5	Merge CB-580: fail a ticket when its member reaches a terminal health state CI / contract (push) Successful in 41s Details CI / build (push) Successful in 1m32s Details Verified by the lead on `0af902e`: mvn -f bridged/pom.xml clean install, unpiped, in a scratch worktree — MVN_EXIT=0, Tests run: 689, Failures: 0, Errors: 0, BUILD SUCCESS. Reviewed by the lead reading the diff. The member's own report was lost to the idle reaper before it was collected, so there was no author write-up to review against. All three defects that sank `3b2f395` are absent: * failTarget is required — one constructor, Objects.requireNonNull, no defaulting overload anywhere in the repo, and Bridged.java:365 updated to pass messages::abandon. * The failure fires inside reportTransition after its `if (previous == next) return;` guard, so an unchanged tick cannot reach it. * No AtomicReference; MessageService is passed directly, so there is no empty window. abandon(String target, String reason) confirmed as CB-568's target-wide operation that resolves waiters as a failure rather than letting them time out. Known limitation, accepted and covered by its own test: states.put records the new state before the bounded retries run, so if all three attempts throw, the tickets stay pending and no later tick retries. It is logged at WARN, and the retries carry no backoff.	2026-08-15 08:57:03 +02:00
ltms	500bfa2c33	Merge CB-576: release preserves a dirty worktree instead of deleting it CI / build (push) Successful in 51s Details CI / contract (push) Successful in 1m1s Details Verified by the lead on `b525b0f`: mvn -f bridged/pom.xml clean install, unpiped, in a scratch worktree — MVN_EXIT=0, Tests run: 686, Failures: 0, Errors: 0, BUILD SUCCESS. Review accepted the required-interface-method shape (no defaulting overload) and the decision to count untracked files as dirty — the work lost in the incident was a file that was never added. One blocking defect was found and fixed in `b525b0f`: hasUncommitted called git with no existence check, so a missing worktree threw WorktreeException from inside release() after registry.remove() but before notifyReleased() and launcher.stop(), orphaning the pane and stranding a blocked bridge_send caller. It now mirrors remove()'s already-gone tolerance. Waived on merge, tracked as follow-up: a SessionManager-level test that teardown completes when the worktree is gone, and the stronger fix behind it — reapIdle calls release() with no try/catch while drainAll wraps it, so any exception in that window aborts the whole reaping pass.	2026-08-15 08:54:33 +02:00
Dai Ha	b525b0f08f	CB-576: hasUncommitted tolerates an already-gone worktree CI / contract (pull_request) Successful in 1m8s Details CI / build (pull_request) Successful in 1m37s Details	2026-08-15 08:47:12 +02:00
Dai Ha	976eff8ad1	CB-579: resolve a lead by its tab name, drop the terminal-id pin CI / build (pull_request) Successful in 52s Details CI / contract (pull_request) Successful in 1m18s Details Leader.terminal -> Leader.tab (exact tab label, case-insensitive match). LeadTabScanner matches an exact tab->name map instead of stripping a shared tabPrefix, and no longer merges configured leads into every scan result -- a stale pin can no longer outlive its tab. LeadLauncher.tabLabel() returns the configured tab directly; the terminalId pinned-terminal fallback in liveLeads() is gone. Config load now rejects a leftover fleet.leaders.*.terminal key instead of silently ignoring it. primary.terminal is untouched.	2026-08-15 08:05:21 +02:00
Dai Ha	0af902ec43	CB-580: fail a ticket when its member reaches a terminal health state CI / build (pull_request) Successful in 54s Details CI / contract (pull_request) Successful in 1m5s Details FleetHealthMonitor now requires a failTarget BiConsumer<String,String> collaborator (no defaulting overload) and calls it exactly once when a member transitions into GONE or NEVER_READY, via CB-568's idempotent target-wide abandon() operation. The reason string names the real terminal state. failTarget invocation retries up to MAX_FAIL_TARGET_ATTEMPTS (3) within the same transition if it throws, and never refires on a later tick where the state is unchanged. Bridged.java wires messages::abandon as the production failTarget.	2026-08-15 07:54:28 +02:00
Dai Ha	9118ce2537	CB-576: release preserves a dirty worktree instead of deleting it CI / contract (pull_request) Successful in 1m2s Details CI / build (pull_request) Successful in 1m35s Details	2026-08-15 07:34:06 +02:00
Dai Ha	2f48e08f1f	Merge CB-577 follow-up: drop the target-keyed async index CI / contract (push) Successful in 43s Details CI / build (push) Successful in 55s Details asyncTasksByWaiter correlates an async question by the exact rendezvous waiter, so the target-keyed set it replaced can no longer decide anything. Keeping it meant two indexes of the same fact, one of them ambiguous whenever a target has two accepted tickets. The race the old test modelled by reflection is gone with it: identity keys make 'some other task reached this target' unrepresentable, so there is no longer a wrong task for the question to land on. Also carries the criterion-1 doc correction, which is identical to `aac29d6` on a different parent.	2026-08-15 06:40:31 +02:00
Dai Ha	fec284e7cb	Merge M4 unit 2a: accepted-turn identity carried to delivery CI / build (push) Successful in 59s Details CI / contract (push) Successful in 1m17s Details TurnToken, owned by MessageService, binds a target to the exact rendezvous waiter for one accepted send. Injector.Pending carries it and the delivery callback hands it to CompletionResolver, so the baseline is bound to the send it belongs to by construction rather than by a lookup that could pick a different one. The callback signature is required, not a defaulted overload: a delivery with no token is exactly the unbound baseline this unit forbids, so a default would let a caller silently produce it. The token deliberately omits the session turn number. MessageService owns acceptance but never learns of delivery, and CompletionResolver.onDelivered runs before SessionManager.onDelivered, so the number does not exist yet at the only point the token could capture it. docs/M4-Fleet-Health.md criterion 1 records this and the two rejected alternatives. Still open for the next slice: the missing/post-restart baseline test and the no-replay test.	2026-08-15 06:36:46 +02:00
Dai Ha	8e2e4c5e73	M4 unit 2a: migrate test call sites to the required turn token The delivery callback now requires a TurnToken, so 54 test call sites had to pass one. They use an explicit TestTurnTokens.inert(target) rather than a defaulted overload, because a delivery with no token is the unbound baseline this unit forbids. The first version of inert() returned a fresh CompletableFuture as the waiter, which turned captureBaselineSkipsTheReadWhenNoSendIsWaiting red: the resolver saw a non-null waiter, concluded a turn was in flight, and scraped a pane no send was blocked on. An inert value must omit the fact, not invent it, so the waiter is now null and the production skip fires as designed.	2026-08-15 06:36:35 +02:00
Dai Ha	b745e159de	CB-573: carry accepted turn tokens on delivery	2026-08-15 06:30:41 +02:00
Dai Ha	aac29d604c	M4: correct unit 2 criterion 1 — TurnToken cannot carry the session turn CI / contract (push) Successful in 1m0s Details CI / build (push) Successful in 1m32s Details The criterion required the token to bind the session turn number. Three independent refusals from the implementer showed why that is not implementable at this layer: MessageService owns acceptance but never learns of delivery, and CompletionResolver.onDelivered runs before SessionManager.onDelivered, so the turn number does not exist yet at the only point the token could capture it. Records both rejected alternatives and why, so the next reader does not re-derive them: a target-keyed registry restores the ambiguity the token exists to remove, and injecting a turn counter couples layers to fill a field nothing reads yet.	2026-08-15 06:29:27 +02:00