fleetd #702: pin the architect-demotion window and split a thin test

Add a CallerResolverTest case proving a releasing architect resolves as
WORKER while still inside releaseRemoved's teardown window, with a
control resolve outside the window that must stay ARCHITECT so the
in-window assertion cannot pass against a slot that was never bound.

Split spawnedMemberRoleSurvivesAnOverlappingReleaseThatUnmarksEarly into
two SessionManagerTest cases, each reaching Releasing.enter/leave
directly through reflection so a mutation to one invariant (the depth
count in leave, or enter's prior-terminal preservation) can only fail
its own test.

CLAUDE.md

@@ -27,10 +27,11 @@ through its `fleet_*` tools. No session addresses a peer, a broker, or the netwo
 **Every role reads this file.** A member runs in a git worktree of this same repo, so it inherits
 this `CLAUDE.md` verbatim, and every rule below is role-conditional.
 
-**Call `fleet_whoami`.** It returns `primary`, `worker`, or `architect`, resolved by the daemon from
-your connection — unforgeable, and the same resolution its authorization gate uses. A worker also
-carries its `sessionId`, `profile`, `worktree` and `branch`; an architect carries the slot name it
-was bound to. Don't infer what you can ask.
+**Call `fleet_whoami`.** It returns `primary`, `worker`, `architect`, or `collaborator`, resolved by
+the daemon from your connection — unforgeable, and the same resolution its authorization gate uses.
+A worker also carries its `sessionId`, `profile`, `worktree` and `branch`; an architect carries the
+slot name it was bound to; a collaborator carries its registry name and its own `sessionId`, and
+**no `leader` key** — a collaborator is a named peer, not a primary. Don't infer what you can ask.
 
 Only if that call is unavailable, fall back to these — each is one-way, so keep reading until one
 fires: the reply charter in your system prompt (*"You are a spawned member in the
@@ -38,12 +39,17 @@ claude-bridge fleet"*) ⇒ **spawned member**; fleet tools prefixed `mcp__fleet_
 member** (the launcher fixes that mount name; a primary's mount is named by whoever wrote its
 `.mcp.json`, so it varies — and a member spawned before CB-632 still says `mcp__bridge__*`); `ANTHROPIC_BASE_URL` set ⇒ **spawned member** (Claude-model members run
 on a clean env, so its *absence* proves nothing). None of these separate a worker from an architect —
-only `fleet_whoami` does. **Still unsure ⇒ act as a worker**, the most restricted member role. The
+only `fleet_whoami` does. **And none of them fires for a collaborator at all**: every signal in the
+ladder detects a *spawned* member, while a collaborator is a tab a person opened by hand, so it has
+no charter, no fixed mount name and a normal environment. A collaborator that cannot call
+`fleet_whoami` therefore falls to the line below and acts as a worker. That is the safe direction —
+it under-privileges, and the refusals are loud — but it means a collaborator has no way to learn
+what it is except by asking. **Still unsure ⇒ act as a worker**, the most restricted member role. The
 two mistakes are not symmetric: a primary acting as a worker is refused by the authorization gate —
 loud and self-correcting — while a member acting as the primary ends its turn with no `fleet_reply`,
 and the sender silently receives nothing. Fail toward the recoverable error.
 
-### Invariants — both roles, no exceptions
+### Invariants — every role, no exceptions
 
 1. **Never set, export, or forward `ANTHROPIC_BASE_URL`** (or `ANTHROPIC_AUTH_TOKEN`). The primary
    stays on subscription; only the bridge puts a member off it, at spawn. Mounting the bridge must
@@ -51,9 +57,10 @@ and the sender silently receives nothing. Fail toward the recoverable error.
 2. **The bridge is the only channel.** Text you print in your terminal reaches nobody — the other
    side cannot see your screen. An answer that isn't in a `fleet_*` call is silently discarded.
 3. **Identity comes from the connection, never an argument.** Workers never pass a target; you
-   cannot act as another session. Spawn/stop/drain are lead-only; **send is lead or architect**;
-   reply/ask are only-as-itself — any peer may answer for its own pane, and for no other. A call
-   outside your role is refused, not queued.
+   cannot act as another session. Spawn/stop/drain are lead-only; **send is lead, architect, or
+   collaborator** — and a collaborator may send only to a lead or another collaborator, never to a
+   spawned member's terminal; reply/ask are only-as-itself — any peer may answer for its own pane,
+   and for no other. A call outside your role is refused, not queued.
 4. **Delivery is status-gated: one message per turn.** Don't busy-poll a peer's terminal and don't
    re-send because a call looks slow — the bridge delivers when the peer is `idle`, `blocked` or
    `done`. A spawned member must **also** have mounted the bridge MCP: until it has, it is not
@@ -93,7 +100,9 @@ below are the procedure — run them in order, every task, not only the big ones
 5. **Collect** — `fleet_poll{ticket}` → `fleet_ack{target, msgId}`. Answer a worker's `fleet_ask`
    with `fleet_send{turnId, content}` — **not** `sessionId`. A worker gone quiet is diagnosed with
    `fleet_status`, never by reading its terminal; it also reports an open question and the `turnId`
-   that answers it. **A worker's ask waits ~55 seconds, and no nudge makes that longer** — so never
+   that answers it — but **only to the caller that created that delegation**, so a question raised
+   under an architect's brief is invisible to you, and seeing none does not mean there is none.
+   **A worker's ask waits ~55 seconds, and no nudge makes that longer** — so never
    brief a worker to "ask me". Decide before you delegate, or give it an explicit default.
    **A correction cannot reach a busy member.** A `fleet_send` to a working member is *accepted* and
    returns a ticket, and is then never delivered — measured here three times in one session, and the
@@ -161,6 +170,7 @@ you decide.
 | Answer a member's `fleet_ask` | `fleet_send{turnId, content}` — **not** `sessionId` |
 | Message a **peer lead** on this host | `fleet_send{sessionId: <their terminal>, content}` — `fleet_list` → `leads` reports it. Coordination only, **never** a task |
 | Message a **peer lead** on another daemon or host | `fleet_send{coordId: <their coord-id>, content}` — needs a `coordinator:` block; your own coord-id is in `fleet_list`. Coordination only, **never** a task |
+| Message a **collaborator** on this host | `fleet_send{sessionId: <their terminal>, content}` — `fleet_list` reports a `collaborators` array, and each row carries that peer's `name` and the `sessionId` you send to. It is visible to you, to an architect and to another collaborator, never to a worker. Coordination only, **never** a task |
 | Answer a peer lead that messaged you | `fleet_send{coordId}` — or `{sessionId}` if they are on this host. **Not** `fleet_reply`: it has no peer route and the publish is refused |
 | Read your own held lead-to-lead mail (no ack) | `fleet_poll{coordId: <your own coord-id, from fleet_list's coordinator.selfId>}` — primary-only; never acks, so `fleet_list`'s `held[]` still shows it after. `fleet_list`'s `held[]` gives only a truncated preview — this is the only way to read the full body |
 | Collect a held reply | `fleet_poll{target}` · then `fleet_ack{target, msgId}` |
@@ -234,6 +244,27 @@ simply complies has thrown away the reason there are two of you.
 7. **Never merge.** Stage files explicitly — never `git add -A` — and leave alone anything the
    project marks as not-yours-to-commit.
 
+### Collaborator — a named peer, not a member
+
+`fleet_whoami` answered `collaborator`, so your pane's tab matches a `fleet.collaborators.<name>.tab`
+entry. You are **not** a member: nothing delegates to you, you have no brief, no worktree and no
+ticket, and **you owe no `fleet_reply`** — the turn contract above is for a session a lead spawned,
+and it does not apply to you. Read it only to understand what the members around you are doing.
+
+What you may do: observe the fleet (`fleet_list`, `fleet_profiles`, `fleet_whoami`), and send to a
+lead or to another collaborator. What you may not: spawn, stop or drain anything, roll a lead's
+session, answer a member's `fleet_ask`, poll a ticket, or send to a spawned member's terminal. Each
+of those is refused at the gate, not queued.
+
+Two limits worth knowing before you hit them. **You cannot reach a worker** — not even to help one —
+because a worker belongs to the lead that spawned it, and routing around that would make you a
+second orchestrator with no plan. Send to the lead instead. And **you cannot read a ticket**, so you
+cannot collect a delegation's reply: `fleet_poll` refuses you at the role gate, and a ticket also
+records the terminal that created it, so even a leaked id reads nothing.
+
+Being named buys you a channel, not authority. Your `fleet_send` to a lead is coordination between
+peers: the lead owes you no obedience, and you owe it none.
+
 ### Where each rule lives (don't duplicate — extend the right layer)
 
 | Layer | Scope | Reaches |
@@ -359,7 +390,7 @@ Before you call any work done, check the row that matches what you touched:
 | `ConnectionIdentity` / how a caller is resolved | the `fleet_whoami` paragraph and the fallback ladder |
 | `REPLY_CHARTER`, or a launcher's mount/flags | the fallback ladder (`mcp__fleet__*`), and the layering table's top row |
 | the injector / status gating | invariant 4 |
-| worktree provisioning or the parity overlay | the "both roles read this file" premise — it rests on the worker's worktree being a checkout of this repo |
+| worktree provisioning or the parity overlay | the "every role reads this file" premise — it rests on the worker's worktree being a checkout of this repo |
 | `.claude/skills/**` | the addendum's skill list, and the "name the playbook" rule |
 | a new peer kind (non-Claude adapter) | what that peer can read — anything it must obey belongs in its charter, not in the block |
 | **anything an operator can use, configure, or observe** — an MCP tool, a `fleetd.yaml` knob, an endpoint, a visible behaviour | **[Features](wiki/11-Features.md)** — one entry: what it does · the knob that turns it on · **why it exists** · the gotcha |

fleetd/fleetd.example.yaml

@@ -62,11 +62,12 @@ bind:
 #
 # Leads are configured under `fleet.leaders:` — see THE FLEET further down.
 #
-# Two things stop the tab-name convention from becoming a way to claim leadership: the configured
-# member spaces are excluded from the scan, so nothing fleetd places can land in a matching tab;
-# and startup REFUSES a `tabPrefix` that the fleet tabLabel template, or any per-profile `tabLabel`
-# override, also matches — so the two namespaces cannot overlap by accident. The label is a NAME,
-# never a capability: what a pane may do is decided by the role the daemon resolves for it.
+# Two things stop the tab-name convention from becoming a way to claim leadership: startup REFUSES
+# a `tabPrefix` that the fleet tabLabel template, or any per-profile `tabLabel` override, also
+# matches, so the two namespaces cannot overlap by accident; and the CallerResolver asks the live
+# spawned-member roster BEFORE any tab map, so a live member is never mistaken for a lead no matter
+# what its tab says. The label is a NAME, never a capability: what a pane may do is decided by the
+# role the daemon resolves for it.
 
 # CB-551: IDLE-LEAD HEARTBEAT — nudge the single lead back to work when it has been continuously
 # idle (no open fleet_send driving it) past the quiet period. The fleet is one lead + architects +
@@ -659,9 +660,10 @@ fleet:
   #     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
-  #                                  # scan, so a lead placed in one is never found again.
+  #     workspace: leads             # where a launched lead's tab is created (default "fleet",
+  #                                  # the same shared space the members use). Sharing that space
+  #                                  # with members is the normal shipped shape: the scanner tells
+  #                                  # a lead from a member by the exact tab label, not by workspace.
   #     cwd: /path/to/repo           # the launched lead's working directory (default: fleetd's own)
   #     kind: claude                 # descriptive; reported by fleet_whoami
   #   gpt-sol-5.6:
@@ -669,11 +671,18 @@ fleet:
   #     kind: opencode
   #     model: openai/gpt-5.6-terra
 
-  # A collaborator tab, keyed by name (fleetd #669). This block is parsed and validated today;
-  # nothing yet recognises or addresses the tab it names. Recognise-only, like a profile-less
-  # `leaders:` entry above: there is no `profile:`, no `instances:` and no `kind:`. `tab:` is
-  # REQUIRED and is the only field identity depends on, matched case-insensitively — the same
-  # GET-THE-VALUE-RIGHT warning above the `leaders:` block applies here too.
+  # A collaborator tab, keyed by name (fleetd #669). A pane whose tab matches resolves as the
+  # COLLABORATOR role: `fleet_whoami` answers `collaborator`, and the session may observe the fleet
+  # (`fleet_list`, `fleet_profiles`, `fleet_whoami`) and `fleet_send` to a lead or to another
+  # collaborator. It may NOT spawn, stop or drain anything, roll a lead's session, answer a
+  # member's `fleet_ask`, poll a ticket, send across hosts, or send to a spawned member's terminal.
+  # Each of those is refused at the gate rather than queued.
+  # Recognise-only, like a profile-less `leaders:` entry above: there is no `profile:`, no
+  # `instances:` and no `kind:`. `tab:` is REQUIRED and is the only field identity depends on,
+  # matched case-insensitively — the same GET-THE-VALUE-RIGHT warning above the `leaders:` block
+  # applies here too.
+  # A lead cannot discover a collaborator yet (#703): `fleet_list` has no `collaborators` key, so
+  # the collaborator must speak first, or pass on the `sessionId` its own `fleet_whoami` reports.
   # collaborators:
   #   reviewer-alex:
   #     tab: "collab: alex"

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

@@ -217,25 +217,28 @@ public final class Fleetd {
 
     /**
      * The {@link Injector}'s readiness gate (CB-534): a target is deliverable if it is a spawned
-     * member whose agent has connected the bridge MCP, <em>or</em> a lead.
+     * member whose agent has connected the bridge MCP, a lead, <em>or</em> a collaborator.
      *
      * <p>The gate exists for one reason — to hold a delivery out of a <em>spawned</em> member's boot
      * window, where herdr already reports {@code idle} but the TUI would drop an injected paste. That
      * hazard is a property of spawning. A lead is never spawned: the operator started it and named it
-     * (or labelled its tab) only once it was up, so there is no boot window to guard.
+     * (or labelled its tab) only once it was up, so there is no boot window to guard. A collaborator
+     * is the same way — a person's own tab, matched to a configured name, never spawned.
      *
-     * <p>A lead is also never enrolled in {@link MemberPresence} — {@code FleetMcp} marks presence
-     * for every spawned member (worker and architect), deliberately, since that map doubles as the
-     * member roster's availability signal and a lead counted there would show up as an available
-     * member. So without the second disjunct a lead is permanently un-deliverable: every
-     * lead→lead send sat on the gate for {@code READINESS_GRACE_POLLS} (~60s) and then failed
-     * having never been typed into the pane.
+     * <p>Neither a lead nor a collaborator is ever enrolled in {@link MemberPresence} — {@code
+     * FleetMcp} marks presence for every spawned member (worker and architect), deliberately, since
+     * that map doubles as the member roster's availability signal and a lead or collaborator counted
+     * there would show up as an available member. So without the second and third disjuncts a lead or
+     * collaborator is permanently un-deliverable: every send to one sat on the gate for
+     * {@code READINESS_GRACE_POLLS} (~60s) and then failed having never been typed into the pane.
      *
-     * <p>The lead set is read through the supplier on each call rather than snapshotted, so a lead
-     * discovered by {@code leadScan} after startup becomes deliverable without a restart.
+     * <p>Both sets are read through their supplier on each call rather than snapshotted, so a lead or
+     * collaborator discovered by {@code leadScan} after startup becomes deliverable without a restart.
      */
-    static Predicate<String> deliverableTo(MemberPresence presence, Supplier<Map<String, String>> leads) {
-        return target -> presence.isPresent(target) || leads.get().containsKey(target);
+    static Predicate<String> deliverableTo(MemberPresence presence, Supplier<Map<String, String>> leads,
+            Supplier<Map<String, String>> collaborators) {
+        return target -> presence.isPresent(target) || leads.get().containsKey(target)
+                || collaborators.get().containsKey(target);
     }
 
     /**

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

@@ -43,6 +43,7 @@ import dev.ltms.fleet.msg.MessageService;
 import dev.ltms.fleet.msg.Rendezvous;
 import dev.ltms.fleet.msg.ReplyInbox;
 import dev.ltms.fleet.msg.ReplyPushLoop;
+import dev.ltms.fleet.peer.MemberRole;
 import dev.ltms.fleet.peer.PeerLauncher;
 import dev.ltms.fleet.placement.BackendOutagePolicy;
 import dev.ltms.fleet.placement.BackendQuarantine;
@@ -141,8 +142,12 @@ final class FleetdAssembly {
                 ? ports.connectHerdr(Path.of(cfg.memberHerdrSocket()))
                 : herdr;
         AtomicReference<Supplier<Map<String, String>>> leadsRef = new AtomicReference<>(Map::of);
+        // fleetd #669 Unit E: a collaborator's pane is opened by a person, exactly like a lead's,
+        // so its terminal must also route to the lead herdr daemon rather than the member one.
+        AtomicReference<Supplier<Map<String, String>>> collaboratorTerminalsRef = new AtomicReference<>(Map::of);
         HerdrRouter router = new HerdrRouter(herdr, memberHerdr,
-                target -> leadsRef.get().get().containsKey(target));
+                target -> leadsRef.get().get().containsKey(target)
+                        || collaboratorTerminalsRef.get().get().containsKey(target));
         // CB-402: one adapter per configured peer kind, fronted by a composite router. A profile's
         // `kind:` selects its adapter — claude-code (the default) and opencode partition the profile
         // set — and the composite dispatches each SPI call to the adapter that owns the profile/pane.
@@ -250,26 +255,47 @@ final class FleetdAssembly {
             log.info("leads: {} panes recognised {}", leadTerminals.size(), leadTerminals.values());
         }
         // CB-531/CB-579: discover leads by the tab labels the operator writes, one scanner per
-        // configured lead's own exact `tab:` label.
+        // configured lead's own exact `tab:` label. fleetd #669: the same scan also recognises a
+        // configured collaborator's tab, so one herdr pass answers both.
         final Supplier<Map<String, String>> leads;
+        final Supplier<Map<String, String>> collaboratorTerminals;
         var leaders = cfg.fleet().leaders();
-        if (!leaders.isEmpty()) {
+        var collaboratorsConfig = cfg.fleet().collaborators();
+        if (!leaders.isEmpty() || !collaboratorsConfig.isEmpty()) {
             Map<String, String> tabToName = new LinkedHashMap<>();
             leaders.forEach((name, leader) -> {
                 if (leader != null && leader.tab() != null && !leader.tab().isBlank()) {
                     tabToName.put(leader.tab(), name);
                 }
             });
-            int scanIntervalSeconds = leaders.values().iterator().next().scanIntervalSeconds();
+            Map<String, String> collaboratorTabToName = new LinkedHashMap<>();
+            collaboratorsConfig.forEach((name, collaborator) -> {
+                if (collaborator != null && collaborator.tab() != null && !collaborator.tab().isBlank()) {
+                    collaboratorTabToName.put(collaborator.tab(), name);
+                }
+            });
+            // A collaborator-only fleet configures no `leaders:` entry to read a scan interval from
+            // — FleetConfig.Collaborator carries no scanIntervalSeconds of its own. Falling back to
+            // FleetConfig.Leader's own compact-constructor default keeps a collaborator-only
+            // deployment on the same rescan cadence as the default lead cadence, instead of
+            // inventing a second number for the same kind of scan.
+            int scanIntervalSeconds = leaders.isEmpty()
+                    ? 10
+                    : leaders.values().iterator().next().scanIntervalSeconds();
             // This must use the lead daemon: scanning member tabs would demote the lead to a worker.
-            leads = new LeadTabScanner(herdr, tabToName, Set.of(),
+            LeadTabScanner scanner = new LeadTabScanner(herdr, tabToName, collaboratorTabToName, Set.of(),
                     TimeUnit.SECONDS.toNanos(scanIntervalSeconds), ports.nanoClock());
-            log.info("lead scan: tabs {} host a lead (rescan every {}s, shared fleet space)",
-                    tabToName.keySet(), scanIntervalSeconds);
+            leads = scanner;
+            collaboratorTerminals = scanner::collaborators;
+            log.info("lead/collaborator scan: tabs {} host a lead, tabs {} host a collaborator "
+                            + "(rescan every {}s, shared fleet space)",
+                    tabToName.keySet(), collaboratorTabToName.keySet(), scanIntervalSeconds);
         } else {
             leads = () -> leadTerminals;
+            collaboratorTerminals = Map::of;
         }
         leadsRef.set(leads);
+        collaboratorTerminalsRef.set(collaboratorTerminals);
 
         // CB-558: start any declared lead that is not already running. After the scanner is built,
         // and only when herdr answered — the launcher's whole safety property is that it can count
@@ -341,7 +367,7 @@ final class FleetdAssembly {
         // CB-301: the manager's presence bridge records availability and drives SPAWNING → READY.
         MemberPresence presence = sessions.asPresence();
         TurnListener turnListener = Fleetd.turnListener(completion, sessions);
-        Predicate<String> deliverable = Fleetd.deliverableTo(presence, leads);
+        Predicate<String> deliverable = Fleetd.deliverableTo(presence, leads, collaboratorTerminals);
         // fleetd #556: registration is wired directly to `completion`, not folded into the
         // `turnListener` fan-out above — so it survives `sessions.onDelivered` (or any future
         // listener) throwing, regardless of call order.
@@ -452,6 +478,11 @@ final class FleetdAssembly {
         ConnectionIdentity identity = new ConnectionIdentity(
                 new PaneLocator(herdr, memberHerdr), new LsofPeerPidLookup(), new LsofProcessCwdLookup());
 
+        // fleetd #669 Unit D: a live spawned member resolves as its own role, whatever a tab map
+        // says about the same terminal. fleetd #702: SessionManager.spawnedMemberRole also answers
+        // for a pane mid-teardown, not only one still in the registry — see its javadoc.
+        Function<String, MemberRole> spawnedMemberRole = sessions::spawnedMemberRole;
+
         // CB-501: one resolver behind both entry paths. Worker identity still comes from the
         // connection and is never token-gated, so enabling token mode cannot lock the fleet out.
         final CallerResolver callers;
@@ -461,11 +492,13 @@ final class FleetdAssembly {
                 throw new IllegalStateException("auth.mode=token but env var " + cfg.auth().tokenEnv()
                         + " is unset or empty — export it before starting fleetd");
             }
-            callers = CallerResolver.withLeadsAndMembers(identity, true, token, leads, members);
+            callers = CallerResolver.withLeadsAndMembers(identity, true, token, leads, members,
+                    spawnedMemberRole, collaboratorTerminals);
             log.info("auth: token mode (bearer required for non-worker callers, env {})",
                     cfg.auth().tokenEnv());
         } else {
-            callers = CallerResolver.withLeadsAndMembers(identity, false, null, leads, members);
+            callers = CallerResolver.withLeadsAndMembers(identity, false, null, leads, members,
+                    spawnedMemberRole, collaboratorTerminals);
             log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
         }
 

fleetd/src/main/java/dev/ltms/fleet/auth/Authz.java

@@ -62,11 +62,11 @@ public final class Authz {
     }
 
     /**
-     * Stands in for the terminal-to-tab registry a collaborator's {@code SEND} is checked
-     * against, until one exists: answers no for every target, so a collaborator reaches nothing
-     * today. Both production gates ({@code FleetMcp#denyFor}, {@code FleetApp#allow}) pass this
-     * exact instance, so the classifier is defined once and replacing it is a one-line change in
-     * each.
+     * The fail-closed classifier: answers no for every target, so a collaborator's {@code SEND}
+     * is refused unless a caller supplies a real one. {@code CallerResolver#knownLeadOrCollaborator()}
+     * is the real one, read from the same lead and collaborator maps {@code CallerResolver#resolve}
+     * consults, so a target that classifier calls known is one {@code resolve} would actually
+     * resolve as a lead or collaborator.
      */
     public static final Predicate<String> NO_KNOWN_LEAD_OR_COLLABORATOR = target -> false;
 
@@ -75,6 +75,9 @@ public final class Authz {
      * {@code SEND} is refused, as if no terminal were a configured lead or collaborator — the
      * same decision {@link #NO_KNOWN_LEAD_OR_COLLABORATOR} gives explicitly. Every other action's
      * result is identical to the four-argument form's, since none of them consult the classifier.
+     *
+     * <p>Its default classifier denies every collaborator, so a caller enforcing authorization
+     * must use the four-argument form instead.
      */
     public static boolean permits(Principal caller, Action action, String targetSession) {
         return permits(caller, action, targetSession, NO_KNOWN_LEAD_OR_COLLABORATOR);

fleetd/src/main/java/dev/ltms/fleet/auth/CallerResolver.java

@@ -7,6 +7,7 @@ import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.util.Map;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /**
@@ -19,6 +20,13 @@ import java.util.function.Supplier;
  *
  * <p><strong>Resolution order</strong> — connection identity first, token second, nothing third:
  * <ol>
+ *   <li>A loopback peer PID that maps to a pane this gateway itself spawned ⇒ that member's own
+ *       role: {@link Role#WORKER} for a dev, hunter, or reviewer; {@link Role#ARCHITECT} for an
+ *       architect, but only while the live slot role still confirms it (fleetd #424 — a slot
+ *       revoked from config demotes an already-bound session on its very next request, so the
+ *       roster's own role is never granted on its word alone). No tab map is consulted — a live
+ *       spawned member's identity comes from the registry that spawned it, never from a label a
+ *       pane could also carry.</li>
  *   <li>A loopback peer PID that maps to a pane named by {@code leaders:}, by the legacy
  *       {@code primary.terminal} pin, or by an operator-labelled lead tab (CB-307, CB-530, CB-531)
  *       ⇒ {@link Role#PRIMARY}, carrying that lead's
@@ -28,8 +36,10 @@ import java.util.function.Supplier;
  *       so two leads can work as peers rather than one being demoted.</li>
  *   <li>A loopback peer PID that maps to a pane bound to a CB-548 architect slot ⇒
  *       {@link Role#ARCHITECT}, carrying the slot name. Just unforgeable as a worker's, and
- *       resolved from the <em>live</em> terminal→slot binding (never a request argument), before
- *       the generic worker fallback.</li>
+ *       resolved from the <em>live</em> terminal→slot binding (never a request argument). This is
+ *       the case the previous step does not catch: a binding with no live spawned-member session.</li>
+ *   <li>A loopback peer PID that maps to an operator-labelled collaborator tab ⇒
+ *       {@link Role#COLLABORATOR}, carrying that collaborator's name.</li>
  *   <li>A loopback peer PID that maps to any other herdr pane ⇒ {@link Role#WORKER}. This is
  *       unforgeable (the OS reports the PID, herdr owns the PID→pane map) and is honoured
  *       regardless of auth mode, so enabling auth never breaks the fleet.</li>
@@ -64,6 +74,20 @@ public final class CallerResolver {
     private final Supplier<Map<String, String>> architectTerminals;
     private final Function<String, MemberRole> memberSlotRoles;
     private final Function<String, String> memberSlotNames;
+    /**
+     * terminal_id → the role of the live spawned member occupying it, or {@code null} for a
+     * terminal no spawned member occupies. Consulted first, ahead of every tab map: a live
+     * spawned member's identity is its own, whatever a tab map says about the same terminal.
+     * A function rather than the roster itself, so a resolve on the hot path never scans a list —
+     * the lookup strategy is the caller's to choose.
+     */
+    private final Function<String, MemberRole> spawnedMemberRole;
+    /**
+     * terminal_id → collaborator name; empty when none are configured. A supplier for the same
+     * reason as {@link #leadTerminals}: a collaborator tab recognised after construction (the tab
+     * scan discovering a newly-labelled tab) takes effect without a restart.
+     */
+    private final Supplier<Map<String, String>> collaboratorTerminals;
 
     /** Loopback-trust resolver: no token required, historical behaviour. Test-only. */
     CallerResolver(ConnectionIdentity identity) {
@@ -124,17 +148,42 @@ public final class CallerResolver {
     /**
      * Live registry form that can confirm a bound slot is an architect slot.
      *
-     * <p>This is the only public construction path. It keeps terminal bindings and slot roles in
-     * the same {@link MemberRegistry}, so a configured architect can resolve as an architect.
+     * <p>It keeps terminal bindings and slot roles in the same {@link MemberRegistry}, so a
+     * configured architect can resolve as an architect. No spawned-member roster or collaborator
+     * registry is consulted — equivalent to {@link #withLeadsAndMembers(ConnectionIdentity,
+     * boolean, String, Supplier, MemberRegistry, Function, Supplier)} with both absent. Kept for
+     * every caller that has neither to offer, so adding them did not churn every construction site.
      */
     public static CallerResolver withLeadsAndMembers(ConnectionIdentity identity,
                                                        boolean tokenMode, String token,
                                                        Supplier<Map<String, String>> leadTerminals,
                                                        MemberRegistry members) {
+        return withLeadsAndMembers(identity, tokenMode, token, leadTerminals, members, null, null);
+    }
+
+    /**
+     * Live registry form that also resolves a live spawned member to its own role, and a
+     * configured collaborator tab to {@link Role#COLLABORATOR}.
+     *
+     * <p>This is the only public construction path that exercises the full resolution order.
+     *
+     * @param spawnedMemberRole      terminal_id → the role of the live spawned member occupying
+     *                               it, or {@code null} for a terminal no spawned member occupies.
+     *                               {@code null} here means no roster is consulted at all (every
+     *                               terminal falls through to the tab maps), not that none matches.
+     * @param collaboratorTerminals  terminal_id → collaborator name, live like {@code leadTerminals}
+     */
+    public static CallerResolver withLeadsAndMembers(ConnectionIdentity identity,
+                                                       boolean tokenMode, String token,
+                                                       Supplier<Map<String, String>> leadTerminals,
+                                                       MemberRegistry members,
+                                                       Function<String, MemberRole> spawnedMemberRole,
+                                                       Supplier<Map<String, String>> collaboratorTerminals) {
         return new CallerResolver(identity, tokenMode, token, leadTerminals,
                 members == null ? null : members::snapshot,
                 members == null ? null : members::roleForSlot,
-                members == null ? null : members::nameForSlot);
+                members == null ? null : members::nameForSlot,
+                spawnedMemberRole, collaboratorTerminals);
     }
 
     private static Supplier<Map<String, String>> fixed(Map<String, String> leadTerminals) {
@@ -160,6 +209,17 @@ public final class CallerResolver {
                            Supplier<Map<String, String>> architectTerminals,
                            Function<String, MemberRole> memberSlotRoles,
                            Function<String, String> memberSlotNames) {
+        this(identity, tokenMode, token, leadTerminals, architectTerminals, memberSlotRoles,
+                memberSlotNames, null, null);
+    }
+
+    private CallerResolver(ConnectionIdentity identity, boolean tokenMode, String token,
+                           Supplier<Map<String, String>> leadTerminals,
+                           Supplier<Map<String, String>> architectTerminals,
+                           Function<String, MemberRole> memberSlotRoles,
+                           Function<String, String> memberSlotNames,
+                           Function<String, MemberRole> spawnedMemberRole,
+                           Supplier<Map<String, String>> collaboratorTerminals) {
         if (tokenMode && (token == null || token.isBlank())) {
             throw new IllegalArgumentException(
                     "auth.mode=token requires a non-empty token; check that the env var named by "
@@ -172,6 +232,8 @@ public final class CallerResolver {
         this.architectTerminals = architectTerminals == null ? Map::of : architectTerminals;
         this.memberSlotRoles = memberSlotRoles == null ? _ -> null : memberSlotRoles;
         this.memberSlotNames = memberSlotNames == null ? Function.identity() : memberSlotNames;
+        this.spawnedMemberRole = spawnedMemberRole == null ? _ -> null : spawnedMemberRole;
+        this.collaboratorTerminals = collaboratorTerminals == null ? Map::of : collaboratorTerminals;
     }
 
     /**
@@ -188,14 +250,24 @@ public final class CallerResolver {
     }
 
     /**
-     * The currently-recognised architect slots, {@code terminal_id → slot name} (CB-548).
+     * The currently-recognised collaborator tabs, {@code terminal_id → name}.
      *
-     * <p>Read from the same supplier {@link #resolve} consults, so a slot that is <em>listed</em>
-     * here but would not <em>resolve</em> (or the reverse) cannot drift apart. Live for the same
-     * reason as {@link #leads()}.
+     * <p>Read from the same supplier {@link #resolve} consults, for the reason given in
+     * {@link #leads()}. Live for the same reason as {@link #leads()}.
      */
-    public Map<String, String> members() {
-        return architectTerminals.get();
+    public Map<String, String> collaborators() {
+        return collaboratorTerminals.get();
+    }
+
+    /**
+     * Whether {@code target} names a terminal this resolver would resolve as a lead or a
+     * collaborator — the classifier a collaborator's {@code SEND} is checked against, read from the
+     * exact maps {@link #resolve} consults so a target that would resolve as a lead or collaborator
+     * is never the one a collaborator is refused to reach, or the reverse.
+     */
+    public Predicate<String> knownLeadOrCollaborator() {
+        return target -> leadTerminals.get().containsKey(target)
+                || collaboratorTerminals.get().containsKey(target);
     }
 
     /**
@@ -208,6 +280,25 @@ public final class CallerResolver {
     public Principal resolve(String remoteAddr, int remotePort, String authorizationHeader) {
         ConnectionIdentity.Caller c = identity.resolve(remoteAddr, remotePort);
         if (c.terminal() != null) {
+            MemberRole spawnedRole = spawnedMemberRole.apply(c.terminal());
+            if (spawnedRole != null) {
+                // A live spawned member occupies this pane. Its identity is its own, whatever a tab
+                // map says about the same terminal — checked before every tab map, consulting none
+                // of them, so a tab label can never override a roster entry for the same terminal.
+                if (spawnedRole == MemberRole.ARCHITECT) {
+                    // The roster only answers THAT this pane is a live spawned member; config still
+                    // decides WHAT that member's slot grants (fleetd #424). A slot revoked after the
+                    // bind must still demote this session on its very next request, so the roster's
+                    // own ARCHITECT role is confirmed against the live slot role, exactly as the
+                    // architect-slot step below confirms a binding with no live member session.
+                    String slot = architectTerminals.get().get(c.terminal());
+                    if (slot != null && memberSlotRoles.apply(slot) == MemberRole.ARCHITECT) {
+                        return Principal.architect(memberSlotNames.apply(slot), c.terminal(), c.pid());
+                    }
+                    return Principal.worker(c.terminal(), c.pid());
+                }
+                return Principal.worker(c.terminal(), c.pid());
+            }
             String lead = leadTerminals.get().get(c.terminal());
             if (lead != null) {
                 // The config names this pane as a lead's own. The pane mapping is exactly as
@@ -221,10 +312,17 @@ public final class CallerResolver {
                 // The config/live binding names this pane as an architect slot's own. Same
                 // unforgeable pane mapping; the live binding, never a request argument, decides.
                 // Check the slot role too: this defence in depth prevents a bad lifecycle bind from
-                // escalating a dev, hunter or reviewer into an architect. Checked before
-                // the worker fallback.
+                // escalating a dev, hunter or reviewer into an architect. This is the case the
+                // spawned-member step above does not catch: a binding with no live member session.
                 return Principal.architect(memberSlotNames.apply(slot), c.terminal(), c.pid());
             }
+            String collaborator = collaboratorTerminals.get().get(c.terminal());
+            if (collaborator != null) {
+                // An operator-labelled collaborator tab, confirmed live by the same scan that
+                // confirms a lead tab. Checked last among the tab maps so a pane also matching one
+                // of the above keeps that stronger role.
+                return Principal.collaborator(collaborator, c.terminal(), c.pid());
+            }
             return Principal.worker(c.terminal(), c.pid()); // unforgeable; never token-gated
         }
 

fleetd/src/main/java/dev/ltms/fleet/config/ConfigRef.java

@@ -615,7 +615,7 @@ public final class ConfigRef implements Supplier<FleetConfig> {
         // may bind to, AND what a slot already bound still grants) through its own instance of that
         // same supplier shape — see MemberRegistry.live and its class doc for the binding rule:
         // removing a slot revokes ARCHITECT on the bound pane's very next request, and only the slot
-        // OCCUPANCY survives, so the demoted session keeps its slot key until it unbinds. Only
+        // OCCUPANCY survives, so the demoted session keeps its slot key until it unbinds.
         // fleet.leaders is frozen (Fleetd.java:281 reads cfg.fleet().leaders() off the startup
         // snapshot to build both the LeadTabScanner's tab-label-to-name map, wired into
         // CallerResolver.withLeadsAndMembers at Fleetd.java:620/624, and — when herdr answered —
@@ -635,6 +635,11 @@ public final class ConfigRef implements Supplier<FleetConfig> {
                     + "live through that same supplier for placement AND through a separate supplier "
                     + "on MemberRegistry for spawn-time identity — both already applied");
         }
+        if (!Objects.equals(collaboratorsOf(old), collaboratorsOf(fresh))) {
+            changed.add("fleet: fleet.collaborators (each collaborator's tab) is read once at "
+                    + "startup to build the LeadTabScanner's identity map, which is not rebuilt on "
+                    + "reload, so a collaborator added, removed, or given a new tab: needs a restart");
+        }
         // Kept in step with SPLIT_KEYS the same way changedColdKeys is kept in step with COLD_KEYS —
         // every message here must be traceable to one of the split keys the class doc documents.
         // NOTE what this does NOT prove, per the javadoc above: it does not catch a SPLIT_KEYS
@@ -650,6 +655,11 @@ public final class ConfigRef implements Supplier<FleetConfig> {
         return cfg.fleet() == null ? Map.of() : cfg.fleet().leaders();
     }
 
+    /** {@code cfg.fleet().collaborators()}, defensively, in case a caller hands in a non-defaulted config. */
+    private static Map<String, FleetConfig.Collaborator> collaboratorsOf(FleetConfig cfg) {
+        return cfg.fleet() == null ? Map.of() : cfg.fleet().collaborators();
+    }
+
     /**
      * {@link FleetConfig.Profile} record components deliberately left out of
      * {@link #sameLaunchSettings} because they are read <em>live</em>, not baked in at spawn — see

fleetd/src/main/java/dev/ltms/fleet/config/FleetConfig.java

@@ -2776,9 +2776,10 @@ public record FleetConfig(
         });
         if (!bad.isEmpty()) {
             throw new IllegalStateException("refusing to start: " + String.join("; ", bad)
-                    + ". Every member labelled that way would be read back as a lead or "
-                    + "collaborator and granted that identity's authority. Change one of the two "
-                    + "so member tabs cannot be confused with a lead's or collaborator's tab.");
+                    + ". A member labelled that way, while its pane carries no entry in the "
+                    + "spawned-member roster, is read back as a lead or collaborator and granted "
+                    + "that identity's authority. Change one of the two so member tabs cannot be "
+                    + "confused with a lead's or collaborator's tab.");
         }
 
         List<String> collisions = new ArrayList<>();
@@ -2877,7 +2878,8 @@ public record FleetConfig(
      * the focused tab rather than its own, so it can land inside a lead's or collaborator's own
      * labelled tab. {@link dev.ltms.fleet.herdr.LeadTabScanner} identifies a lead or collaborator
      * purely by that tab's label — it does not exclude the member space — so a member that ends up
-     * there would be read back as that lead or collaborator and granted that identity's authority.
+     * there, while its pane carries no entry in the spawned-member roster, is read back as that
+     * lead or collaborator and granted that identity's authority.
      *
      * <p>Only an entry with a non-blank {@code tab} is in scope: one with no {@code tab} feeds
      * nothing into {@link dev.ltms.fleet.herdr.LeadTabScanner}, so it creates no hazard here.
@@ -2908,8 +2910,9 @@ public record FleetConfig(
         }
         throw new IllegalStateException("refusing to start: profile(s) " + bad
                 + " use placement: pane while fleet.leaders or fleet.collaborators names a tab. A "
-                + "pane-placed member can land inside that labelled tab and be read back as the "
-                + "lead or collaborator, granted that identity's authority. Set placement: tab for "
+                + "pane-placed member can land inside that labelled tab, and while its pane "
+                + "carries no entry in the spawned-member roster, it is read back as the lead or "
+                + "collaborator and granted that identity's authority. Set placement: tab for "
                 + "each named profile, or remove the tab from every fleet.leaders and "
                 + "fleet.collaborators entry.");
     }

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

@@ -11,12 +11,18 @@ public final class HerdrRouter implements AutoCloseable {
     private final AgentControl memberAgents;
     private final WorkspaceControl leadSpaces;
     private final WorkspaceControl memberSpaces;
-    private final Predicate<String> isLead;
+    private final Predicate<String> routeToLead;
 
-    public HerdrRouter(HerdrClient lead, HerdrClient member, Predicate<String> isLead) {
+    /**
+     * @param routeToLead true for a terminal whose pane lives in the lead herdr daemon — a lead's
+     *                     own pane or a configured collaborator's, both opened by a person at a
+     *                     terminal rather than spawned, so both are found in the lead daemon rather
+     *                     than the member one
+     */
+    public HerdrRouter(HerdrClient lead, HerdrClient member, Predicate<String> routeToLead) {
         this.lead = Objects.requireNonNull(lead, "lead");
         this.member = member != null ? member : lead;
-        this.isLead = Objects.requireNonNull(isLead, "isLead");
+        this.routeToLead = Objects.requireNonNull(routeToLead, "routeToLead");
         leadAgents = new AgentControl(this.lead);
         memberAgents = this.member == this.lead ? leadAgents : new AgentControl(this.member);
         leadSpaces = new WorkspaceControl(this.lead);
@@ -27,7 +33,7 @@ public final class HerdrRouter implements AutoCloseable {
     public WorkspaceControl leadSpaces() { return leadSpaces; }
     public AgentControl memberAgents() { return memberAgents; }
     public WorkspaceControl memberSpaces() { return memberSpaces; }
-    public AgentControl agentsFor(String targetId) { return isLead.test(targetId) ? leadAgents : memberAgents; }
+    public AgentControl agentsFor(String targetId) { return routeToLead.test(targetId) ? leadAgents : memberAgents; }
 
     HerdrClient leadClient() { return lead; }
     HerdrClient memberClient() { return member; }

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

@@ -96,13 +96,19 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
 
     private static final Logger log = LoggerFactory.getLogger(LeadTabScanner.class);
 
+    /** What a matched tab names: a lead or a collaborator. */
+    private enum Kind { LEAD, COLLABORATOR }
+
+    /** One matched tab's name and what it names. */
+    private record Entry(String name, Kind kind) {}
+
     private final HerdrClient herdr;
-    private final Map<String, String> tabToName;
+    private final Map<String, Entry> tabToEntry;
     private final Set<String> excludedWorkspaceLabels;
     private final long ttlNanos;
     private final LongSupplier clock;
 
-    private Map<String, String> cached = Map.of();
+    private Map<String, Entry> cached = Map.of();
     private long scannedAtNanos;
     private boolean everScanned;
 
@@ -126,26 +132,53 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
      */
     public LeadTabScanner(HerdrClient herdr, Map<String, String> tabToName,
                           Set<String> excludedWorkspaceLabels, long ttlNanos, LongSupplier clock) {
+        this(herdr, tabToName, Map.of(), excludedWorkspaceLabels, ttlNanos, clock);
+    }
+
+    /**
+     * As {@link #LeadTabScanner(HerdrClient, Map, Set, long, LongSupplier)}, additionally scanning
+     * for configured collaborator tabs in the same pass.
+     *
+     * @param collaboratorTabToName every configured collaborator's exact tab label → its name
+     *                              ({@code fleet.collaborators.<name>.tab}), matched the same way as
+     *                              {@code tabToName}
+     */
+    public LeadTabScanner(HerdrClient herdr, Map<String, String> tabToName,
+                          Map<String, String> collaboratorTabToName,
+                          Set<String> excludedWorkspaceLabels, long ttlNanos, LongSupplier clock) {
         this.herdr = herdr;
-        this.tabToName = normalize(tabToName);
+        this.tabToEntry = buildTabIndex(tabToName, collaboratorTabToName);
         this.excludedWorkspaceLabels = excludedWorkspaceLabels == null
                 ? Set.of() : Set.copyOf(excludedWorkspaceLabels);
         this.ttlNanos = ttlNanos;
         this.clock = clock;
     }
 
-    /** 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();
+    /**
+     * Keys stripped and lower-cased once, so every lookup is a plain map hit. Leads and
+     * collaborators merge into a single index, so {@link #scan()} matches both kinds in one pass
+     * over the tab list; a label naming both a lead and a collaborator takes the lead entry —
+     * leads are put last, so a colliding key's lead entry is the one that overwrites — since a lead
+     * can already do everything a collaborator can. Config validation already refuses a lead and a
+     * collaborator sharing one exact tab, so this ordering is defence in depth, not the control.
+     */
+    private static Map<String, Entry> buildTabIndex(Map<String, String> tabToName,
+                                                     Map<String, String> collaboratorTabToName) {
+        Map<String, Entry> out = new LinkedHashMap<>();
+        putNormalized(out, collaboratorTabToName, Kind.COLLABORATOR);
+        putNormalized(out, tabToName, Kind.LEAD);
+        return Collections.unmodifiableMap(out);
+    }
+
+    private static void putNormalized(Map<String, Entry> out, Map<String, String> tabToName, Kind kind) {
+        if (tabToName == null) {
+            return;
         }
-        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);
+                out.put(tab.strip().toLowerCase(Locale.ROOT), new Entry(name, kind));
             }
         });
-        return Collections.unmodifiableMap(out);
     }
 
     /**
@@ -156,6 +189,29 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
      */
     @Override
     public synchronized Map<String, String> get() {
+        return byKind(refresh(), Kind.LEAD);
+    }
+
+    /**
+     * The current {@code terminal_id → collaborator name} map, sharing the same scan and cache as
+     * {@link #get()} — both kinds are matched in one pass, so this never costs a second herdr call.
+     */
+    public synchronized Map<String, String> collaborators() {
+        return byKind(refresh(), Kind.COLLABORATOR);
+    }
+
+    private static Map<String, String> byKind(Map<String, Entry> entries, Kind kind) {
+        Map<String, String> out = new LinkedHashMap<>();
+        entries.forEach((terminal, entry) -> {
+            if (entry.kind() == kind) {
+                out.put(terminal, entry.name());
+            }
+        });
+        return Collections.unmodifiableMap(out);
+    }
+
+    /** Rescans if the cache has expired, otherwise returns the cached answer. */
+    private Map<String, Entry> refresh() {
         long now = clock.getAsLong();
         if (everScanned && now - scannedAtNanos < ttlNanos) {
             return cached;
@@ -165,21 +221,21 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
         scannedAtNanos = now;
         everScanned = true;
         try {
-            Map<String, String> fresh = scan();
+            Map<String, Entry> fresh = scan();
             if (!fresh.equals(cached)) {
-                log.info("lead panes: {}", fresh);
+                log.info("lead/collaborator panes: {}", fresh);
             }
             cached = fresh;
         } catch (HerdrException e) {
-            log.warn("lead-tab scan failed, keeping the {} lead(s) already known: {}",
+            log.warn("lead-tab scan failed, keeping the {} entr(y/ies) already known: {}",
                     cached.size(), e.getMessage());
         }
         return cached;
     }
 
     /** One full pass: labelled tabs → live agents in them → those panes' terminals. */
-    private Map<String, String> scan() {
-        Map<String, String> nameByTab = new LinkedHashMap<>();
+    private Map<String, Entry> scan() {
+        Map<String, Entry> entryByTab = new LinkedHashMap<>();
         for (JsonNode w : herdr.call("workspace.list").path("workspaces")) {
             Workspace ws = Workspace.from(w);
             if (ws.workspaceId() == null || excludedWorkspaceLabels.contains(ws.label())) {
@@ -187,21 +243,22 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
             }
             for (JsonNode t : herdr.call("tab.list", Map.of("workspace_id", ws.workspaceId())).path("tabs")) {
                 Tab tab = Tab.from(t);
-                String name = leadNameOf(tab.label());
-                if (name != null && tab.tabId() != null) {
-                    nameByTab.put(tab.tabId(), name);
+                Entry entry = entryOf(tab.label());
+                if (entry != null && tab.tabId() != null) {
+                    entryByTab.put(tab.tabId(), entry);
                 }
             }
         }
 
-        if (nameByTab.isEmpty()) {
+        if (entryByTab.isEmpty()) {
             gracedTerminals = Set.of();
             return Map.of();
         }
 
-        // fleetd #359: a labelled tab is only a lead when herdr also reports a running agent in
-        // it — the same liveness signal LeadLauncher.countLeads trusts for the identical purpose.
-        // Without this, a tab left behind by a session that has since died reads as live forever.
+        // fleetd #359: a labelled tab is only a lead (or collaborator) when herdr also reports a
+        // running agent in it — the same liveness signal LeadLauncher.countLeads trusts for the
+        // identical purpose. Without this, a tab left behind by a session that has since died reads
+        // as live forever.
         Set<String> tabsWithAgent = new HashSet<>();
         for (JsonNode a : herdr.call("agent.list").path("agents")) {
             String tabId = a.path("tab_id").asText(null);
@@ -210,18 +267,18 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
             }
         }
 
-        Map<String, String> byTerminal = new LinkedHashMap<>();
+        Map<String, Entry> byTerminal = new LinkedHashMap<>();
         Set<String> stillGraced = new HashSet<>();
         // One pane.list for every tab: panes carry tab_id, so the join is local.
         for (JsonNode p : herdr.call("pane.list", Map.of()).path("panes")) {
             String tabId = p.path("tab_id").asText(null);
-            String name = nameByTab.get(tabId);
+            Entry entry = entryByTab.get(tabId);
             String terminal = p.path("terminal_id").asText(null);
-            if (name == null || terminal == null || terminal.isBlank()) {
+            if (entry == null || terminal == null || terminal.isBlank()) {
                 continue;
             }
             if (tabsWithAgent.contains(tabId)) {
-                byTerminal.put(terminal, name);
+                byTerminal.put(terminal, entry);
                 continue;
             }
             // No agent reported for this tab, but its tab/pane are still here — this is the
@@ -230,7 +287,7 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
             // reported as live; a terminal we never reported live gets none, so the original #359
             // fix (a genuinely dead tab is never reported) is unaffected for the common case.
             if (cached.containsKey(terminal) && !gracedTerminals.contains(terminal)) {
-                byTerminal.put(terminal, name);
+                byTerminal.put(terminal, entry);
                 stillGraced.add(terminal);
             }
         }
@@ -239,18 +296,19 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
     }
 
     /**
-     * The lead name a tab label declares, or {@code null} if it names none of the configured leads.
+     * The entry a tab label declares, or {@code null} if it names neither a configured lead nor a
+     * configured collaborator.
      *
-     * <p>Exact match (case-insensitive, ends stripped) against {@link #tabToName} — no prefix
+     * <p>Exact match (case-insensitive, ends stripped) against {@link #tabToEntry} — 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. The match strips a trailing
      * {@link PendingCloseMarker} first, so a tab {@code LeadLauncher} has flagged as maybe-dead but
      * not yet closed keeps resolving normally while that reconcile is pending.
      */
-    private String leadNameOf(String label) {
+    private Entry entryOf(String label) {
         if (label == null) {
             return null;
         }
-        return tabToName.get(PendingCloseMarker.strip(label).toLowerCase(Locale.ROOT));
+        return tabToEntry.get(PendingCloseMarker.strip(label).toLowerCase(Locale.ROOT));
     }
 }

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

@@ -114,6 +114,13 @@ public final class FleetMcp {
      * dev.ltms.fleet.herdr.PaneLocator} — that a real assembly wired up. See {@link #identity()}.
      */
     private final ConnectionIdentity identity;
+    /**
+     * Kept as a field (rather than only captured by the {@code contextExtractor} closure) so
+     * {@link #denyFor} can read {@link CallerResolver#knownLeadOrCollaborator()} — the classifier a
+     * collaborator's {@code SEND} is checked against, built from the same lead and collaborator
+     * maps {@link #identity}-based resolution reads.
+     */
+    private final CallerResolver callers;
     private final Metrics metrics;      // CB-502: null → auth failures not counted
     private final CapacitySource capacity;
     private final HealthCoverageSource healthCoverage;
@@ -409,6 +416,7 @@ public final class FleetMcp {
         this.authorizationEnforced = Objects.requireNonNull(authorizationMode, "authorizationMode")
                 == AuthorizationMode.ENFORCED;
         this.identity = identity;
+        this.callers = callers;
         this.leadChannel = leadChannel;
         this.peers = peers == null ? List.of() : List.copyOf(peers);
         this.capacity = capacity;
@@ -482,7 +490,7 @@ public final class FleetMcp {
                     Runnable onAccepted = () -> primaryRegistry.recordDelegation(target, caller);
                     // wait defaults to true (block for the reply); wait:false is fire-and-poll.
                     return Boolean.FALSE.equals(a.get("wait"))
-                            ? sendAsync(messages, target, content, onAccepted, workers.profiles())
+                            ? sendAsync(messages, target, content, onAccepted, workers.profiles(), caller)
                             : send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles());
                 };
         // fleet_reply's identity is the CONNECTION, never an argument — so the authz check
@@ -506,7 +514,7 @@ public final class FleetMcp {
                 (exchange, req) -> {
                     McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_status", req.arguments()), null);
                     if (denied != null) return denied;
-                    return status(messages, str(req.arguments(), "sessionId"));
+                    return status(messages, str(req.arguments(), "sessionId"), callerTerminal(exchange));
                 };
         BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> pollHandler =
                 (exchange, req) -> {
@@ -516,7 +524,7 @@ public final class FleetMcp {
                     // The action depends on the ARGUMENTS, not on the tool name -- see pollAction.
                     McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_poll", a), target);
                     if (denied != null) return denied;
-                    return poll(messages, leadChannel, str(a, "ticket"), target, coordId);
+                    return poll(messages, leadChannel, str(a, "ticket"), target, coordId, callerTerminal(exchange));
                 };
         // CB-307 Increment 3: per-msgId ack (not needed in v1 but supported by the inbox).
         // Acking removes a reply from the inbox, so it is a drain, not a read.
@@ -552,8 +560,10 @@ public final class FleetMcp {
                     return listFleet(workers, sessions, messages, capacity, healthCoverage, loopHealth, quarantine, outage,
                             leadSeats, leadContextGauge, leadConfigDirs, callers.leads(),
                             callerTerminal(exchange),
+                            callers.collaborators(), collaboratorsVisibleTo(principal(exchange)),
                             new CoordinationSource(leadChannel, peers),
-                            coordinatorVisibleTo(principal(exchange)));
+                            coordinatorVisibleTo(principal(exchange)),
+                            leadsVisibleTo(principal(exchange)), membersVisibleTo(principal(exchange)));
                 };
         BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> stopHandler =
                 (exchange, req) -> {
@@ -689,7 +699,7 @@ public final class FleetMcp {
         if (!authorizationEnforced) {
             return null; // AuthorizationMode.UNENFORCED: authorization not enforced (fleetd #518)
         }
-        if (Authz.permits(caller, action, target, Authz.NO_KNOWN_LEAD_OR_COLLABORATOR)) {
+        if (Authz.permits(caller, action, target, callers.knownLeadOrCollaborator())) {
             if (action != Authz.Action.READ && action != Authz.Action.TASK_READ) {
                 AuditLog.allowed(caller, action, target); // reads would drown the trail
             }
@@ -735,6 +745,48 @@ public final class FleetMcp {
         return caller.isPrimary();
     }
 
+    /**
+     * fleetd #703: who may see {@code fleet_list}'s {@code collaborators} array — a roster of the
+     * human-opened tabs this daemon recognises as named peers. Visible to exactly the roles that
+     * may {@link Authz.Action#SEND} to a named peer ({@code Authz.java}'s {@code SEND} case):
+     * the primary, an architect, and a collaborator (reaching another collaborator or a lead). A
+     * worker holds {@code READ} but can never {@code SEND} to a collaborator, so listing them to a
+     * worker would expose which human tabs exist on the host with no use to that caller. Split out
+     * for the same reason as {@link #coordinatorVisibleTo}: the decision must be unit-testable
+     * without fabricating an SDK {@code McpSyncServerExchange}, and the handler must call this
+     * named predicate rather than inlining the check.
+     */
+    static boolean collaboratorsVisibleTo(Principal caller) {
+        return caller.isPrimary() || caller.isArchitect() || caller.isCollaborator();
+    }
+
+    /**
+     * Who may see {@code fleet_list}'s {@code leads} array — exactly the roles that may
+     * {@link Authz.Action#SEND} to a lead: the primary, an architect, and a collaborator. A
+     * collaborator's own {@code fleet_whoami} carries no lead address, and {@code leads} is the
+     * only place this tool gives one, so a collaborator needs this array to use the send it
+     * already holds. A worker can never {@code SEND} at all, so it still sees neither this array
+     * nor {@code members}; a worker's own facts come from {@code fleet_whoami} instead. Split
+     * out for the same reason as {@link #coordinatorVisibleTo} and
+     * {@link #collaboratorsVisibleTo}: the decision must be unit-testable without fabricating an
+     * SDK {@code McpSyncServerExchange}, and the handler must call this named predicate rather
+     * than inlining the check.
+     */
+    static boolean leadsVisibleTo(Principal caller) {
+        return caller.isPrimary() || caller.isArchitect() || caller.isCollaborator();
+    }
+
+    /**
+     * Who may see {@code fleet_list}'s {@code members} array — the primary and an architect,
+     * which may {@link Authz.Action#SEND} to a member. A worker can never {@code SEND} at all,
+     * and a collaborator may {@code SEND} only to a lead or another collaborator, never to a
+     * spawned member, so neither sees this array even though {@link #leadsVisibleTo} grants a
+     * collaborator the sibling one.
+     */
+    static boolean membersVisibleTo(Principal caller) {
+        return caller.isPrimary() || caller.isArchitect();
+    }
+
     /** The worker identity resolved from this call's connection, or {@code null} if the primary. */
     private static String callerTerminal(McpSyncServerExchange exchange) {
         Object v = exchange.transportContext().get(CALLER_TERMINAL);
@@ -948,6 +1000,17 @@ public final class FleetMcp {
      */
     static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content,
                                               Runnable onAccepted, Set<String> profiles) {
+        return sendAsync(messages, sessionId, content, onAccepted, profiles, null);
+    }
+
+    /**
+     * As above, recording {@code creatorTerminal} as this ticket's owner (the caller's own
+     * terminal, resolved from the connection) so a later {@code fleet_poll{ticket}} only hands the
+     * result back to that same caller — see {@link MessageService#poll(String, String)}.
+     */
+    static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content,
+                                              Runnable onAccepted, Set<String> profiles,
+                                              String creatorTerminal) {
         if (isBlank(sessionId) || isBlank(content)) {
             return error("sessionId and content are required");
         }
@@ -955,7 +1018,7 @@ public final class FleetMcp {
         if (targetError != null) {
             return targetError;
         }
-        String ticket = messages.sendAsync(sessionId, content, onAccepted);
+        String ticket = messages.sendAsync(sessionId, content, onAccepted, creatorTerminal);
         return text("accepted — task delegated. Poll fleet_poll with ticket=" + ticket);
     }
 
@@ -1134,9 +1197,24 @@ public final class FleetMcp {
      * exclusively of {@code ticket}/{@code target} — see {@link #pollAction}'s javadoc for why this
      * is a different inbox (this daemon's own {@link LeadChannel}) that authorizes differently
      * ({@link Authz.Action#COORD_READ}, primary-only) from either of the original two branches.
+     *
+     * <p>Does not check who owns {@code ticket} — see the overload that takes {@code
+     * callerTerminal} for that. Callers that do not resolve a caller terminal (tests, or a surface
+     * with no connection-based identity) use this one.
      */
     static McpSchema.CallToolResult poll(MessageService messages, LeadChannel leadChannel, String ticket,
                                          String target, String coordId) {
+        return poll(messages, leadChannel, ticket, target, coordId, null);
+    }
+
+    /**
+     * As above, refusing a ticket lookup whose caller's terminal differs from the terminal that
+     * created it — see {@link MessageService#poll(String, String)}. {@code callerTerminal} is the
+     * CALLING session's terminal id, resolved by the MCP layer from the connection, never a
+     * client-supplied value.
+     */
+    static McpSchema.CallToolResult poll(MessageService messages, LeadChannel leadChannel, String ticket,
+                                         String target, String coordId, String callerTerminal) {
         if (!isBlank(coordId)) {
             return pollHeldPeerMail(leadChannel, coordId);
         }
@@ -1150,7 +1228,7 @@ public final class FleetMcp {
         if (isBlank(ticket)) {
             return error("ticket (or target) is required");
         }
-        MessageService.TaskView v = messages.poll(ticket);
+        MessageService.TaskView v = messages.poll(ticket, callerTerminal);
         if (v == null) {
             return error("unknown ticket: " + ticket + " (never issued, or expired)");
         }
@@ -1257,16 +1335,20 @@ public final class FleetMcp {
 
     /**
      * {@code fleet_status}: the live lifecycle status of a worker session, plus — when the worker
-     * is paused mid-turn in an async {@code fleet_ask} (CB-582) — the open question and how to
-     * answer it, so a lead on its normal poll cadence does not need the ticket to notice.
+     * is paused mid-turn in an async {@code fleet_ask} — the open question and how to answer it, so
+     * a lead on its normal poll cadence does not need the ticket to notice. The question, its
+     * {@code turnId} and its ticket id are shown only to the caller whose terminal created that
+     * delegation, or to a caller with no terminal at all (the unnamed primary); any other caller
+     * still sees the base status. {@code callerTerminal} is the CALLING session's terminal id,
+     * resolved by the MCP layer from the connection, never a client-supplied value.
      */
-    static McpSchema.CallToolResult status(MessageService messages, String sessionId) {
+    static McpSchema.CallToolResult status(MessageService messages, String sessionId, String callerTerminal) {
         if (isBlank(sessionId)) {
             return error("sessionId is required");
         }
         try {
             String base = messages.status(sessionId).name().toLowerCase();
-            MessageService.PendingAsk ask = messages.pendingAsk(sessionId);
+            MessageService.PendingAsk ask = messages.pendingAsk(sessionId, callerTerminal);
             if (ask == null) {
                 return text(base);
             }
@@ -1753,7 +1835,8 @@ public final class FleetMcp {
                                                Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, loopHealth, quarantine,
-                OutageSource.none(), LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
+                OutageSource.none(), LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, false, true, true);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1762,7 +1845,8 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, OutageSource outage,
                                                Map<String, String> leads, String selfTerm) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, CoordinationSource.none(), false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, CoordinationSource.none(), false, true, true);
     }
 
     /**
@@ -1779,7 +1863,8 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, OutageSource.none(),
-                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, false, true, true);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1788,7 +1873,8 @@ public final class FleetMcp {
                                                QuarantineSource quarantine, OutageSource outage,
                                                Map<String, String> leads, String selfTerm, CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
+                LeadSeatSource.none(), new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, false, true, true);
     }
 
     /**
@@ -1810,7 +1896,8 @@ public final class FleetMcp {
                                                LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
                                                CoordinationSource coordination) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine, outage,
-                leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, false);
+                leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, false, true, true);
     }
 
     /**
@@ -1827,14 +1914,23 @@ public final class FleetMcp {
      *                        {@code false} (fleetd #463: a forgotten argument fails closed, not
      *                        open), so a test that wants the {@code coordinator} row must pass
      *                        an explicit {@code true}
+     * @param leadsVisible    whether this caller may see the {@code leads} array (see
+     *                        {@link #leadsVisibleTo}) — unlike {@code callerIsPrimary}, this is
+     *                        also {@code true} for an architect or a collaborator, so it cannot
+     *                        be derived from {@code callerIsPrimary} alone
+     * @param membersVisible  whether this caller may see the {@code members} array (see
+     *                        {@link #membersVisibleTo}); also {@code true} for an architect, but
+     *                        unlike {@code leadsVisible}, never for a collaborator
      */
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                                 CapacitySource capacity, HealthCoverageSource healthCoverage,
                                                 QuarantineSource quarantine, OutageSource outage,
                                                 LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
-                                                CoordinationSource coordination, boolean callerIsPrimary) {
+                                                CoordinationSource coordination, boolean callerIsPrimary,
+                                                boolean leadsVisible, boolean membersVisible) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine,
-                outage, leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, callerIsPrimary);
+                outage, leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, callerIsPrimary, leadsVisible, membersVisible);
     }
 
     /**
@@ -1844,6 +1940,17 @@ public final class FleetMcp {
      * matter); the one caller that matters for caching, {@code fleet_list}'s MCP handler, passes
      * its own single long-lived instance instead (see {@code FleetMcp}'s {@code leadContextGauge}
      * field).
+     *
+     * @param collaborators        terminal_id → collaborator name, live from the resolver
+     *                              ({@link dev.ltms.fleet.auth.CallerResolver#collaborators()})
+     * @param collaboratorsVisible whether this caller may see the {@code collaborators} array (see
+     *                             {@link #collaboratorsVisibleTo}); every wrapper overload above
+     *                             passes {@code false}, so a test that wants the row must call this
+     *                             overload with an explicit {@code true}
+     * @param leadsVisible         whether this caller may see the {@code leads} array (see
+     *                             {@link #leadsVisibleTo})
+     * @param membersVisible       whether this caller may see the {@code members} array (see
+     *                             {@link #membersVisibleTo})
      */
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                                  CapacitySource capacity, HealthCoverageSource healthCoverage,
@@ -1852,32 +1959,56 @@ public final class FleetMcp {
                                                LeadSeatSource leadSeats, LeadContextGauge contextGauge,
                                                LeadConfigDirSource leadConfigDirs,
                                                Map<String, String> leads, String selfTerm,
-                                               CoordinationSource coordination, boolean callerIsPrimary) {
+                                               Map<String, String> collaborators, boolean collaboratorsVisible,
+                                               CoordinationSource coordination, boolean callerIsPrimary,
+                                               boolean leadsVisible, boolean membersVisible) {
         try {
-            Map<String, Agent> live = workers.list().stream()
-                    .map(Agent.class::cast)
-                    .filter(a -> a.terminalId() != null)
-                    .collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
-            List<Map<String, Object>> leadRows = leads.entrySet().stream()
-                    .sorted(Map.Entry.comparingByValue())
-                    .map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm, contextGauge,
-                            leadConfigDirs))
-                    .toList();
+            // Neither row's assembly (leadView/memberCapacityView probing herdr for live status)
+            // runs unless at least one of them needs the live-agent lookup backing it.
+            Map<String, Agent> live = (leadsVisible || membersVisible)
+                    ? workers.list().stream()
+                            .map(Agent.class::cast)
+                            .filter(a -> a.terminalId() != null)
+                            .collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b))
+                    : Map.of();
             // fleetd #209: this is the caller-driven fleet_list read that actually reports
             // agentSessionId (via memberCapacityView -> SessionManager.rosterView), so it uses the
             // resolving roster; the heartbeat/health/metrics timers stay on the plain sessions.roster().
             List<MemberSession> roster = sessions.rosterResolved();
-            List<Map<String, Object>> out = roster.stream()
-                    .map(s -> memberCapacityView(s, live.get(s.terminalId()), messages, capacity.clock().getAsLong()))
-                    .toList();
             Set<String> profiles = new java.util.TreeSet<>(capacity.configuredProfiles().get());
             roster.stream().map(MemberSession::profile).forEach(profiles::add);
             Map<String, Object> result = new LinkedHashMap<>();
-            result.put("leads", leadRows); result.put("members", out);
+            // READ is permission to enter this tool, not permission to receive every field it can
+            // build -- gate BEFORE assembling each row, so the key is absent rather than
+            // present-and-empty; a caller without either row gets its own facts from fleet_whoami.
+            if (leadsVisible) {
+                List<Map<String, Object>> leadRows = leads.entrySet().stream()
+                        .sorted(Map.Entry.comparingByValue())
+                        .map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm, contextGauge,
+                                leadConfigDirs))
+                        .toList();
+                result.put("leads", leadRows);
+            }
+            if (membersVisible) {
+                List<Map<String, Object>> out = roster.stream()
+                        .map(s -> memberCapacityView(s, live.get(s.terminalId()), messages, capacity.clock().getAsLong()))
+                        .toList();
+                result.put("members", out);
+            }
             result.put("healthCoverage", healthCoverage.value().get());
             result.put("loopHealth", Map.of(
                     "statusPoller", loopHealth.statusPoller().get().name(),
                     "sessionReaper", loopHealth.sessionReaper().get().name()));
+            // fleetd #703: a collaborator tab is a person's own tab, so the row is assembled and
+            // included only for the roles that may SEND to a named peer -- gate BEFORE assembling
+            // it, same reason as the coordinator row just below: the key must be absent for a
+            // worker, never present-and-empty.
+            if (collaboratorsVisible) {
+                result.put("collaborators", collaborators.entrySet().stream()
+                        .sorted(Map.Entry.comparingByValue())
+                        .map(e -> collaboratorRow(e.getKey(), e.getValue()))
+                        .toList());
+            }
             // fleetd #439: coordinator/coordinatorView is lead-to-lead coordination state and must
             // never reach a worker or an architect -- gate BEFORE assembling it, not after, so the
             // key is absent rather than present-and-empty.
@@ -2189,6 +2320,20 @@ public final class FleetMcp {
         return m;
     }
 
+    /**
+     * fleetd #703: one row of the {@code collaborators} array — a named peer's registry name and
+     * the herdr {@code terminal_id} a {@code fleet_send} must target to reach it. No status, no
+     * context gauge, no profile: a collaborator is never spawned and carries no profile, so those
+     * fields have no meaning for it, and the scan behind {@code terminal} only reports a tab it
+     * actually found in herdr, so a tab nobody has open does not appear here at all.
+     */
+    private static Map<String, Object> collaboratorRow(String terminal, String name) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("name", name);
+        m.put("sessionId", terminal);
+        return m;
+    }
+
     /**
      * Reads the lead context gauge for one lead. {@code configDir} is this lead's configured
      * {@code CLAUDE_CONFIG_DIR} override (see {@link LeadConfigDirSource}), derived from
@@ -2383,7 +2528,14 @@ public final class FleetMcp {
 
     private static McpSchema.Tool listTool() {
         return tool(FleetTool.LIST.wireName(),
-                "List the whole fleet the bridge tracks, in two parts. 'leads' are your PEERS — other "
+                "List the whole fleet the bridge tracks, in two parts. 'members' is visible to "
+                        + "the primary and an architect only. 'leads' is visible to those two AND a "
+                        + "collaborator — exactly the roles that may fleet_send to a lead, so a "
+                        + "collaborator can learn a lead's sessionId before using the send it already "
+                        + "holds. A worker holds READ to call this tool at all, but gets neither "
+                        + "array, never an empty one; a worker reads its own session, "
+                        + "profile, state, worktree, branch and owner from fleet_whoami instead. "
+                        + "'leads' are your PEERS — other "
                         + "orchestrators, each with its sessionId (the address to fleet_send to), "
                         + "name, live status, and 'self': true on your own row; this is how you "
                         + "discover a peer lead without being told its address. 'members' are the "
@@ -2414,7 +2566,12 @@ public final class FleetMcp {
                         + "msgId/from/preview, never the full body), and one row per coordinator.peers "
                         + "coord-id ('peers': coordId/reachable, plus pending/consumers when reachable) — "
                         + "this is peer DISCOVERY for cross-host leads, distinct from the local 'leads' "
-                        + "array above. It is omitted entirely when no coordinator is configured.",
+                        + "array above. It is omitted entirely when no coordinator is configured. A "
+                        + "'collaborators' array, visible only to the primary, an architect, and a "
+                        + "collaborator (never a worker), reports every other named peer tab this daemon "
+                        + "recognises: each row is 'name' (its registry name) and 'sessionId' (the "
+                        + "terminal id a fleet_send targets to reach it). Absent entirely for a worker, "
+                        + "whatever is configured.",
                 objectSchema(Map.of(), List.of()));
     }
 

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

@@ -275,10 +275,19 @@ public final class MessageService {
          * {@link #abandon}) can never match again regardless of this flag's value.
          */
         private volatile boolean askTimedOut;
+        /**
+         * The terminal of the caller whose {@code fleet_send{wait:false}} created this ticket, or
+         * {@code null} when that caller had no terminal (the unnamed primary) or the ticket was
+         * created through an overload that does not record one. {@link #poll(String, String)}
+         * compares a polling caller's own terminal against this field before handing back the
+         * ticket's state.
+         */
+        private final String creatorTerminal;
 
-        private Task(String ticket, String target, LongSupplier nowNanos) {
+        private Task(String ticket, String target, LongSupplier nowNanos, String creatorTerminal) {
             this.ticket = ticket;
             this.target = target;
+            this.creatorTerminal = creatorTerminal;
             this.createdNanos = nowNanos.getAsLong();
             future.whenComplete((reply, ex) -> completedNanos = nowNanos.getAsLong());
         }
@@ -1279,8 +1288,20 @@ public final class MessageService {
      * @return the ticket to poll for the eventual result
      */
     public String sendAsync(String target, String content, Runnable onAccepted) {
+        return sendAsync(target, content, onAccepted, null);
+    }
+
+    /**
+     * As {@link #sendAsync(String, String, Runnable)}, recording {@code creatorTerminal} as this
+     * ticket's owner. {@link #poll(String, String)} refuses a later caller whose own terminal
+     * differs from this one; {@code null} records no owner (a caller with no terminal — the
+     * unnamed primary — is always allowed to poll the result regardless).
+     *
+     * @return the ticket to poll for the eventual result
+     */
+    public String sendAsync(String target, String content, Runnable onAccepted, String creatorTerminal) {
         String ticket = "task-" + ticketSeq.incrementAndGet();
-        Task task = new Task(ticket, target, nowNanos);
+        Task task = new Task(ticket, target, nowNanos, creatorTerminal);
         tasks.put(ticket, task);
         if (pushLoop != null) {
             // CB-588: task.future only ever completes on a terminal phase (DONE or a failure) — a
@@ -1343,15 +1364,31 @@ public final class MessageService {
     }
 
     /**
-     * Snapshot the state of an async delegation. Returns {@code null} for an unknown/expired ticket;
-     * otherwise a {@link Phase#PENDING} view (with the live worker status as detail), a
-     * {@link Phase#DONE} view carrying the reply, or a {@link Phase#FAILED} view with the reason.
+     * As {@link #poll(String, String)}, with no caller terminal — the ticket's ownership is never
+     * checked, so this overload must only be used where the caller's identity is otherwise
+     * irrelevant (a test, or a surface that does not resolve a caller terminal at all).
      */
     public TaskView poll(String ticket) {
+        return poll(ticket, null);
+    }
+
+    /**
+     * Snapshot the state of an async delegation. Returns {@code null} for an unknown/expired ticket.
+     * Refuses a {@code callerTerminal} that differs from the terminal that created the ticket (see
+     * {@link #sendAsync(String, String, Runnable, String)}) with a {@link Phase#FAILED} view that
+     * carries no reply text — a caller with no terminal (the unnamed primary) is never refused.
+     * Otherwise returns a {@link Phase#PENDING} view (with the live worker status as detail), a
+     * {@link Phase#DONE} view carrying the reply, or a {@link Phase#FAILED} view with the reason.
+     */
+    public TaskView poll(String ticket, String callerTerminal) {
         Task task = tasks.get(ticket);
         if (task == null) {
             return null;
         }
+        if (!ownsTicket(task, callerTerminal)) {
+            return new TaskView(ticket, Phase.FAILED, null, null,
+                    "forbidden: this ticket was created by a different session", null);
+        }
         CompletableFuture<Reply> f = task.future;
         if (!f.isDone()) {
             Reply question = task.question;
@@ -1387,6 +1424,17 @@ public final class MessageService {
         return new TaskView(ticket, Phase.FAILED, null, null, detail, null);
     }
 
+    /**
+     * Whether {@code callerTerminal} may read {@code task}'s state. A caller with no terminal
+     * always may — that is the unnamed primary, resolved by token or loopback trust, which never
+     * carries a herdr pane and must keep reading every ticket. Otherwise the caller's terminal must
+     * equal the terminal recorded on the task; a task with no recorded terminal matches no
+     * terminal-bearing caller.
+     */
+    private static boolean ownsTicket(Task task, String callerTerminal) {
+        return callerTerminal == null || callerTerminal.equals(task.creatorTerminal);
+    }
+
     /**
      * Test seam only — carries no production behaviour, and nothing in this class calls it;
      * {@link #pruneTerminalTickets} still reads {@link Task#completedNanos} directly.
@@ -1706,16 +1754,17 @@ public final class MessageService {
     }
 
     /**
-     * The question {@code workerSession} is currently paused on via {@code fleet_ask}, if any
-     * (CB-582) — {@code fleet_status} uses this to show a pending question without the caller
-     * needing the ticket. {@code null} when the session has no open async question (including a
-     * session mid a <em>blocking</em> {@code fleet_ask}, which has no {@link Task} to look up — see
-     * {@link PendingAsk}).
+     * The question {@code workerSession} is currently paused on via {@code fleet_ask}, if any —
+     * {@code fleet_status} uses this to show a pending question without the caller needing the
+     * ticket. {@code null} when the session has no open async question (including a session mid a
+     * <em>blocking</em> {@code fleet_ask}, which has no {@link Task} to look up — see
+     * {@link PendingAsk}), or when {@code callerTerminal} does not own the task the question
+     * belongs to (see {@link #ownsTicket(Task, String)}).
      */
-    public PendingAsk pendingAsk(String workerSession) {
+    public PendingAsk pendingAsk(String workerSession, String callerTerminal) {
         for (Task task : tasks.values()) {
             Reply q = task.question;
-            if (q != null && workerSession.equals(task.target)) {
+            if (q != null && workerSession.equals(task.target) && ownsTicket(task, callerTerminal)) {
                 return new PendingAsk(task.ticket, q.text(), q.turnId());
             }
         }

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

@@ -269,12 +269,16 @@ public final class FleetApp {
     /**
      * The authorization decision behind {@link #allow}, taking the caller directly rather than
      * pulling it from a servlet {@link Context} — unit-testable without fabricating a live
-     * request, the same reason {@code FleetMcp#denyFor} is split from {@code FleetMcp#deny}. The
-     * classifier is the same shared instance {@link #allow} would use, so a test calling this
-     * exercises the real production gate, not a test-supplied stand-in.
+     * request, the same reason {@code FleetMcp#denyFor} is split from {@code FleetMcp#deny}.
+     *
+     * @param knownLeadOrCollaborator the classifier a collaborator's {@code SEND} is checked
+     *                                against; pass {@link #auth}'s own {@code
+     *                                knownLeadOrCollaborator()} to exercise the real production
+     *                                gate, as {@link #allow} does
      */
-    static boolean permitsFor(Principal caller, Authz.Action action, String target) {
-        return Authz.permits(caller, action, target, Authz.NO_KNOWN_LEAD_OR_COLLABORATOR);
+    static boolean permitsFor(Principal caller, Authz.Action action, String target,
+                              Predicate<String> knownLeadOrCollaborator) {
+        return Authz.permits(caller, action, target, knownLeadOrCollaborator);
     }
 
     /**
@@ -290,7 +294,7 @@ public final class FleetApp {
             return true; // legacy: authorization not enforced
         }
         Principal caller = ctx.attribute(CALLER);
-        if (permitsFor(caller, action, target)) {
+        if (permitsFor(caller, action, target, auth.knownLeadOrCollaborator())) {
             if (action != Authz.Action.READ && action != Authz.Action.METRICS
                     && action != Authz.Action.TASK_READ) {
                 AuditLog.allowed(caller, action, target); // reads would drown the trail
@@ -690,7 +694,8 @@ public final class FleetApp {
 
         if (!wait) {
             // Fire-and-poll (CB-107): return a ticket immediately; the caller polls GET /tasks/{ticket}.
-            String ticket = messages.sendAsync(id, content);
+            Principal caller = ctx.attribute(CALLER);
+            String ticket = messages.sendAsync(id, content, null, caller == null ? null : caller.terminal());
             ctx.status(202).json(Map.of("sessionId", id, "ticket", ticket, "status", "accepted"));
             return;
         }
@@ -870,10 +875,12 @@ public final class FleetApp {
             body.put("sessionId", id);
             body.put("status", messages.status(id).name().toLowerCase());
             body.put("ready", deliverable.test(id));
-            // CB-582: a worker paused mid-turn in an async fleet_ask is otherwise invisible to a
-            // status poll — surface the open question and how to answer it, same as fleet_poll's
-            // Phase.ASKING view.
-            MessageService.PendingAsk ask = messages.pendingAsk(id);
+            // A worker paused mid-turn in an async fleet_ask is otherwise invisible to a status
+            // poll — surface the open question and how to answer it, same as fleet_poll's
+            // Phase.ASKING view, but only to the caller whose terminal created that delegation, or
+            // to a caller with no terminal at all (the unnamed primary).
+            Principal caller = ctx.attribute(CALLER);
+            MessageService.PendingAsk ask = messages.pendingAsk(id, caller == null ? null : caller.terminal());
             if (ask != null) {
                 body.put("question", ask.question());
                 body.put("turnId", ask.turnId());
@@ -890,7 +897,8 @@ public final class FleetApp {
         if (!allow(ctx, routeAction("GET /tasks/{ticket}"), null)) {
             return;
         }
-        MessageService.TaskView v = messages.poll(ctx.pathParam("ticket"));
+        Principal caller = ctx.attribute(CALLER);
+        MessageService.TaskView v = messages.poll(ctx.pathParam("ticket"), caller == null ? null : caller.terminal());
         if (v == null) {
             ctx.status(404).json(Map.of("error", "unknown_ticket", "detail", "no such task (or it has expired)"));
             return;

fleetd/src/main/java/dev/ltms/fleet/session/SessionManager.java

@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
+import java.util.function.Supplier;
 
 /**
  * Authoritative in-daemon registry of the worker sessions this {@code fleetd} process spawned.
@@ -57,6 +58,18 @@ public final class SessionManager implements TurnListener {
      * Populated on every spawn path, removed on {@link #release}.
      */
     private final ConcurrentHashMap<String /*paneId*/, PeerHandle> handles = new ConcurrentHashMap<>();
+    /**
+     * fleetd #702: a pane mid-teardown, keyed by paneId, held from just before its registry entry
+     * is removed until {@link #releaseRemoved} finishes. {@link #spawnedMemberRole} consults this
+     * alongside the registry, so a caller resolving the pane's terminal during that window still
+     * sees a live member and never falls through to a tab map.
+     *
+     * <p>Depth-counted rather than a plain set: two threads can be tearing down the same pane at
+     * once (the CAS in {@link #releaseIfCurrent} exists for exactly that race), and with a set the
+     * loser's {@code finally} would unmark the pane while the winner is still mid-teardown,
+     * reopening the window this exists to close.
+     */
+    private final ConcurrentHashMap<String /*paneId*/, Releasing> releasing = new ConcurrentHashMap<>();
     private final MemberPresence presence;
     private final SecureRandom nonceRandom = new SecureRandom();
     private final AtomicLong nonceSeq = new AtomicLong();
@@ -303,9 +316,11 @@ public final class SessionManager implements TurnListener {
      * is a logged path an operator can reclaim, the cost of a deleted one is unrecoverable work.
      */
     private MemberSession release(String paneId, ReleaseCause cause) {
-        MemberSession removed = registry.remove(paneId);
-        releaseRemoved(paneId, removed, handles.remove(paneId), cause);
-        return removed;
+        return releaseWindow(paneId, registry.get(paneId), () -> {
+            MemberSession removed = registry.remove(paneId);
+            releaseRemoved(paneId, removed, handles.remove(paneId), cause);
+            return removed;
+        });
     }
 
     /**
@@ -314,16 +329,85 @@ public final class SessionManager implements TurnListener {
      * DONE record from stopping a worker that delivery has made BUSY.
      */
     private boolean releaseIfCurrent(MemberSession expected, ReleaseCause cause) {
-        if (!registry.remove(expected.paneId(), expected)) {
-            // A lifecycle transition replaced the record between the caller's check and this remove.
-            // Log it: this race is by definition unobservable otherwise, and a reaper that silently
-            // declines to reap is the hardest kind of behaviour to diagnose after the fact.
-            log.debug("skipping reap of pane={}: its registry record changed after the idle check "
-                    + "(most likely a delivery made it BUSY)", expected.paneId());
-            return false;
+        return releaseWindow(expected.paneId(), expected, () -> {
+            if (!registry.remove(expected.paneId(), expected)) {
+                // A lifecycle transition replaced the record between the caller's check and this
+                // remove. Log it: this race is by definition unobservable otherwise, and a reaper
+                // that silently declines to reap is the hardest kind of behaviour to diagnose
+                // after the fact.
+                log.debug("skipping reap of pane={}: its registry record changed after the idle "
+                        + "check (most likely a delivery made it BUSY)", expected.paneId());
+                return false;
+            }
+            releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
+            return true;
+        });
+    }
+
+    /**
+     * fleetd #702: mark {@code paneId} as mid-teardown — using {@code known}'s terminal/role when
+     * it is available — for the whole of {@code teardown}, which removes the registry entry and
+     * then runs {@link #releaseRemoved}. Shared by both registry-removal sites ({@link #release}'s
+     * unconditional remove and {@link #releaseIfCurrent}'s CAS remove) so neither can leave the
+     * other's window unmarked.
+     *
+     * <p>The mark is written before {@code teardown} runs — so it covers the removal itself, not
+     * only what comes after it — and cleared in a {@code finally}, so an unchecked throw out of
+     * {@code teardown} (including one from {@link PeerLauncher#stop}, which declares nothing) can
+     * never leave the pane marked for the rest of the daemon's life.
+     */
+    private <T> T releaseWindow(String paneId, MemberSession known, Supplier<T> teardown) {
+        releasing.compute(paneId, (_, prior) -> Releasing.enter(prior, known));
+        try {
+            return teardown.get();
+        } finally {
+            releasing.compute(paneId, (_, prior) -> prior == null ? null : prior.leave());
         }
-        releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
-        return true;
+    }
+
+    /**
+     * Depth count plus the terminal/role a mid-teardown pane belongs to, for
+     * {@link #spawnedMemberRole}. The terminal/role come from whichever call into
+     * {@link #releaseWindow} first knew them — a losing {@link #releaseIfCurrent} CAS has no
+     * {@code known} session of its own, so it must not blank out what the winner already recorded.
+     */
+    private record Releasing(int depth, String terminalId, MemberRole role) {
+        private static Releasing enter(Releasing prior, MemberSession known) {
+            int depth = (prior == null ? 0 : prior.depth()) + 1;
+            String terminalId = known != null ? known.terminalId() : prior == null ? null : prior.terminalId();
+            MemberRole role = known != null ? known.role() : prior == null ? null : prior.role();
+            return new Releasing(depth, terminalId, role);
+        }
+
+        private Releasing leave() {
+            return depth <= 1 ? null : new Releasing(depth - 1, terminalId, role);
+        }
+    }
+
+    /**
+     * The role of the live spawned member occupying {@code terminal} — whether it is currently in
+     * the registry, or mid-teardown between {@link #release} removing its registry entry and
+     * {@link #releaseRemoved} actually stopping its pane (fleetd #702). {@code null} for a terminal
+     * that is neither: this method is the one reader a caller resolver consults before any tab
+     * map, so a live or releasing member's identity never falls back to a tab label.
+     *
+     * <p>Checks the registry directly via {@link #findByTerminal} rather than {@link #roster()},
+     * so this hot-path lookup (consulted on every resolve) never pays for a list copy or a stream.
+     */
+    public MemberRole spawnedMemberRole(String terminal) {
+        MemberSession session = findByTerminal(terminal);
+        if (session != null) {
+            return session.role();
+        }
+        if (terminal == null) {
+            return null;
+        }
+        for (Releasing r : releasing.values()) {
+            if (terminal.equals(r.terminalId())) {
+                return r.role();
+            }
+        }
+        return null;
     }
 
     private void releaseRemoved(String paneId, MemberSession removed, PeerHandle removedHandle,

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

@@ -14,10 +14,12 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * CB-534: the injector's readiness gate must open for a lead as well as for a present worker.
+ * fleetd #669 follow-up: the same gate must also open for a collaborator, which — like a lead —
+ * is never enrolled in {@link MemberPresence} and never discovered by the lead scan.
  *
- * <p>The bug these cover was silent and slow: a lead was never marked present (only workers are), so
- * every lead→lead delivery sat on the gate for the full readiness grace and failed ~60s later without
- * a keystroke ever reaching the pane.
+ * <p>The bug these cover was silent and slow: a lead (and later a collaborator) was never marked
+ * present (only workers are) and never counted as a lead, so every send to one sat on the gate for
+ * the full readiness grace and failed ~60s later without a keystroke ever reaching the pane.
  */
 class FleetDeliverabilityTest {
 
@@ -25,19 +27,25 @@ class FleetDeliverabilityTest {
         return () -> m;
     }
 
+    private static Supplier<Map<String, String>> collaborators(Map<String, String> m) {
+        return () -> m;
+    }
+
     @Test
     @DisplayName("a worker that has connected its MCP is deliverable")
     void presentWorkerIsDeliverable() {
         MemberPresence presence = new MemberPresence();
         presence.markPresent("term_worker");
 
-        assertTrue(Fleetd.deliverableTo(presence, leads(Map.of())).test("term_worker"));
+        assertTrue(Fleetd.deliverableTo(presence, leads(Map.of()), collaborators(Map.of()))
+                .test("term_worker"));
     }
 
     @Test
     @DisplayName("a worker still in its boot window is held back")
     void absentWorkerIsNotDeliverable() {
-        assertFalse(Fleetd.deliverableTo(new MemberPresence(), leads(Map.of())).test("term_booting"));
+        assertFalse(Fleetd.deliverableTo(new MemberPresence(), leads(Map.of()), collaborators(Map.of()))
+                .test("term_booting"));
     }
 
     @Test
@@ -45,19 +53,31 @@ class FleetDeliverabilityTest {
     void leadIsDeliverableWithoutPresence() {
         MemberPresence presence = new MemberPresence();
         Predicate<String> deliverable =
-                Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")));
+                Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")), collaborators(Map.of()));
 
         assertFalse(presence.isPresent("term_lead"), "a lead is never enrolled in worker presence");
         assertTrue(deliverable.test("term_lead"), "…and must be deliverable anyway");
     }
 
     @Test
-    @DisplayName("an unknown terminal is deliverable to neither")
+    @DisplayName("a collaborator is deliverable without ever being marked present or scanned as a lead")
+    void collaboratorIsDeliverableWithoutPresenceOrLeadStatus() {
+        MemberPresence presence = new MemberPresence();
+        Predicate<String> deliverable = Fleetd.deliverableTo(presence, leads(Map.of()),
+                collaborators(Map.of("term_collab", "kevin")));
+
+        assertFalse(presence.isPresent("term_collab"), "a collaborator is never enrolled in worker presence");
+        assertTrue(deliverable.test("term_collab"), "…and must be deliverable anyway");
+    }
+
+    @Test
+    @DisplayName("an unknown terminal is deliverable to none of presence, leads, or collaborators")
     void strangerIsNotDeliverable() {
         MemberPresence presence = new MemberPresence();
         presence.markPresent("term_worker");
 
-        assertFalse(Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")))
+        assertFalse(Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")),
+                        collaborators(Map.of("term_collab", "kevin")))
                 .test("term_stranger"));
     }
 
@@ -65,22 +85,47 @@ class FleetDeliverabilityTest {
     @DisplayName("a lead discovered after startup becomes deliverable with no restart")
     void leadSetIsReadThroughOnEveryCall() {
         Map<String, String> discovered = new HashMap<>();
-        Predicate<String> deliverable = Fleetd.deliverableTo(new MemberPresence(), leads(discovered));
+        Predicate<String> deliverable =
+                Fleetd.deliverableTo(new MemberPresence(), leads(discovered), collaborators(Map.of()));
 
         assertFalse(deliverable.test("term_late"));
         discovered.put("term_late", "gpt-sol-5.6"); // leadScan picks up a newly labelled tab
         assertTrue(deliverable.test("term_late"), "the supplier must be re-read, not snapshotted");
     }
 
+    @Test
+    @DisplayName("a collaborator discovered after startup becomes deliverable with no restart")
+    void collaboratorSetIsReadThroughOnEveryCall() {
+        Map<String, String> discovered = new HashMap<>();
+        Predicate<String> deliverable =
+                Fleetd.deliverableTo(new MemberPresence(), leads(Map.of()), collaborators(discovered));
+
+        assertFalse(deliverable.test("term_late_collab"));
+        discovered.put("term_late_collab", "kevin"); // the same tab scan picks up a newly labelled collaborator tab
+        assertTrue(deliverable.test("term_late_collab"), "the supplier must be re-read, not snapshotted");
+    }
+
     @Test
     @DisplayName("forgetting a torn-down worker does not strip a lead of its deliverability")
     void forgetDoesNotDisarmALead() {
         MemberPresence presence = new MemberPresence();
         Predicate<String> deliverable =
-                Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")));
+                Fleetd.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")), collaborators(Map.of()));
 
         presence.forget("term_lead"); // the injector's cleanup path runs against every target
 
         assertTrue(deliverable.test("term_lead"));
     }
+
+    @Test
+    @DisplayName("forgetting a torn-down worker does not strip a collaborator of its deliverability")
+    void forgetDoesNotDisarmACollaborator() {
+        MemberPresence presence = new MemberPresence();
+        Predicate<String> deliverable = Fleetd.deliverableTo(presence, leads(Map.of()),
+                collaborators(Map.of("term_collab", "kevin")));
+
+        presence.forget("term_collab"); // the injector's cleanup path runs against every target
+
+        assertTrue(deliverable.test("term_collab"));
+    }
 }

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

@@ -0,0 +1,165 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.msg.ReplyInbox;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * fleetd #669 Unit E. Reaches the real {@link dev.ltms.fleet.herdr.HerdrRouter} that {@link
+ * FleetdAssembly#assembleAndStart} builds and wires — not a copy built for this test — and proves
+ * that a configured collaborator's terminal routes to the LEAD herdr daemon.
+ *
+ * <p>Two distinct {@link FakeHerdr} instances are required, the same pattern {@code
+ * FleetdAssemblyConnectionIdentityTest} and {@code FleetdLeadRolloverAssemblyTest} already use:
+ * with one client shared between {@code herdrSocket} and {@code memberHerdrSocket},
+ * {@code HerdrRouter} folds {@code leadAgents} and {@code memberAgents} into the same instance
+ * (see its constructor), and {@code agentsFor} would return that one object regardless of whether
+ * the collaborator map was ever consulted — invisible to a mutation of the predicate this ticket
+ * fixes. This test's two sockets resolve to two different fakes, so the assertion only passes when
+ * the collaborator's terminal is actually recognised and routed to the lead one.
+ */
+class FleetdAssemblyCollaboratorHerdrRoutingTest {
+
+    private static final Path LEAD_SOCKET = Path.of("/fake/lead-herdr.sock");
+    private static final Path MEMBER_SOCKET = Path.of("/fake/member-herdr.sock");
+
+    private static final class RecordingResourcePorts implements ResourcePorts {
+        final Map<Path, HerdrClient> herdrsBySocket = new LinkedHashMap<>();
+        Runnable shutdownHook;
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of();
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            HerdrClient client = herdrsBySocket.get(socketPath);
+            if (client == null) {
+                throw new IllegalStateException("no fake herdr registered for socket " + socketPath);
+            }
+            return client;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> new ReplyInbox() {
+                @Override public void own(String target) { }
+                @Override public void release(String target) { }
+                @Override public void publish(String target, String msgId, String content) { }
+                @Override public List<InboxMessage> peek(String target) { return List.of(); }
+                @Override public boolean ack(String target, String msgId) { return false; }
+            };
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> {
+                throw new UnsupportedOperationException(
+                        "leadMailboxOpener must not be called — no coordinator: block is configured");
+            };
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+            // Do not bind a real port in this assembly test.
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            return () -> {
+                throw new UnsupportedOperationException("FakeHerdr is healthy; no poll wait is expected");
+            };
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: "%s"
+                memberHerdrSocket: "%s"
+                idleSleepGuard:
+                  enabled: false
+                fleet:
+                  collaborators:
+                    reviewer-alex:
+                      tab: "collab: alex"
+                profiles:
+                  sonnet:
+                    subscription: true
+                    argv: ["ccs", "sonnet"]
+                """.formatted(LEAD_SOCKET, MEMBER_SOCKET));
+        return FleetConfig.load(file);
+    }
+
+    @Test
+    void assembledRouterRoutesACollaboratorTerminalToTheLeadDaemon(@TempDir Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir);
+        RecordingResourcePorts ports = new RecordingResourcePorts();
+
+        // The fixed FakeHerdr fixture already ties terminal "term_a" to a live agent on tab
+        // "w2:t7" (pane "w2:p7") — seeding only the tab LABEL to match the configured collaborator
+        // is enough to make LeadTabScanner resolve "term_a" as that collaborator. Seeded on the
+        // LEAD fake only: a collaborator's pane lives in the lead daemon, exactly like a lead's.
+        FakeHerdr lead = new FakeHerdr().withTab("w2", "w2:t7", "collab: alex");
+        FakeHerdr member = new FakeHerdr();
+        ports.herdrsBySocket.put(LEAD_SOCKET, lead);
+        ports.herdrsBySocket.put(MEMBER_SOCKET, member);
+
+        FleetdRuntime runtime = FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg,
+                new ConfigRef(dir.resolve("fleetd.yaml"), cfg), new SubscriptionGuard(cfg.guard().hostSet())), ports);
+        try {
+            assertSame(runtime.router().leadAgents(), runtime.router().agentsFor("term_a"),
+                    "a configured collaborator's terminal must route to the LEAD daemon — "
+                            + "FleetdAssembly must wire the collaborator map into the router's "
+                            + "predicate, not just LeadTabScanner.get()");
+            assertSame(runtime.router().memberAgents(), runtime.router().agentsFor("term_shell"),
+                    "control: a terminal naming neither a lead nor a collaborator (term_shell, on "
+                            + "the unlabelled tab w2:t8) must still route to the member daemon");
+        } finally {
+            assertNotNull(ports.shutdownHook, "control: assembly must capture its shutdown hook");
+            ports.shutdownHook.run();
+        }
+    }
+}

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

@@ -31,8 +31,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * fleetd #670 — pins the {@code excludedWorkspaceLabels} argument {@link FleetdAssembly}'s
- * production boot path passes to {@link LeadTabScanner} at {@code FleetdAssembly.java:265}
- * ({@code Set.of()}).
+ * production boot path passes to {@link LeadTabScanner} ({@code Set.of()}).
  *
  * <p>{@code LeadTabScannerTest} already covers this constructor parameter, but it builds its own
  * {@link LeadTabScanner} with its own set, so it tests the seam and proves nothing about the
@@ -191,7 +190,7 @@ class FleetdAssemblyLeadTabScannerExclusionTest {
             Set<?> excluded = (Set<?>) excludedField.get(leads);
 
             assertTrue(excluded.isEmpty(),
-                    "FleetdAssembly.java:265 must pass an empty excludedWorkspaceLabels to "
+                    "FleetdAssembly must pass an empty excludedWorkspaceLabels to "
                             + "LeadTabScanner — scanning member tabs would demote the lead to a worker");
         } finally {
             assertNotNull(ports.shutdownHook, "control: assembly must capture its shutdown hook");

fleetd/src/test/java/dev/ltms/fleet/auth/CallerResolverTest.java

@@ -1,13 +1,25 @@
 package dev.ltms.fleet.auth;
 
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.FakeHerdr;
 import dev.ltms.fleet.herdr.PaneLocator;
+import dev.ltms.fleet.herdr.WorkspaceControl;
 import dev.ltms.fleet.mcp.ConnectionIdentity;
 import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.member.ClaudeCodeLauncher;
 import dev.ltms.fleet.peer.MemberRole;
+import dev.ltms.fleet.session.MemberSession;
+import dev.ltms.fleet.session.SessionManager;
+import dev.ltms.fleet.session.WorktreeRequest;
+import dev.ltms.fleet.session.Worktrees;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -460,7 +472,6 @@ class CallerResolverTest {
         assertTrue(members.bind("architect:lead-designer", "term_a")); // the later lifecycle binds the slot
 
         assertEquals(Role.ARCHITECT, r.resolve("127.0.0.1", 42, null).role());
-        assertEquals("architect:lead-designer", r.members().get("term_a"));
     }
 
     @Test
@@ -523,4 +534,342 @@ class CallerResolverTest {
         }
     }
 
+    // ── fleetd #669 Unit D: a live spawned member outranks every tab map ────────────────────────
+
+    /**
+     * Criterion 1: a terminal present in BOTH the spawned-member roster AND the lead tab map
+     * resolves as its member role, not as a lead — the roster is checked first, consulting no tab
+     * map at all when it matches.
+     */
+    @Test
+    void aSpawnedMemberWinsOverALeadTabForTheSamePane() {
+        Principal p = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null,
+                        () -> Map.of("term_a", "opus-5.0"), new MemberRegistry(null),
+                        t -> "term_a".equals(t) ? MemberRole.DEV : null, Map::of)
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.WORKER, p.role(),
+                "a live spawned member's own identity must win over a tab map naming the same pane a lead");
+        assertEquals("term_a", p.terminal());
+    }
+
+    /**
+     * fleetd #702: between {@link SessionManager#release} removing the registry entry and the
+     * pane actually stopping, a resolve for that terminal must still see the live member and
+     * never fall through to a tab map — wired through the real {@link SessionManager}, not a
+     * hand-rolled stand-in for its {@code spawnedMemberRole}.
+     *
+     * <p>Reuses the ticket's own test idea: an injected {@code hasUncommitted} resolves the
+     * releasing terminal from inside {@code release}'s window — a real call landing inside the
+     * window, so no sleep and no race.
+     *
+     * <p>The property asserted is "no tab map is consulted", never "the same role is returned".
+     * The mandatory control is the lead tab map: it names this exact terminal, and the same
+     * resolve taken <em>outside</em> the window (before release runs) must still return the lead
+     * role — without that control, the in-window assertion would also pass on an empty map and
+     * prove nothing.
+     */
+    @Test
+    void aPaneMidTeardownResolvesAsItsOwnRoleConsultingNoTabMap() {
+        FakeHerdr herdr = new FakeHerdr().pinNextStarts(1, "term_a", "w2:p7");
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
+                List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
+
+        AtomicReference<Principal> duringWindow = new AtomicReference<>();
+        AtomicReference<CallerResolver> resolverRef = new AtomicReference<>();
+        Worktrees worktrees = new Worktrees() {
+            @Override
+            public String add(String repoRoot, String branch, String baseRef) {
+                return "/wt/" + branch.replace('/', '_');
+            }
+
+            @Override
+            public void remove(String repoRoot, String worktreePath) {
+            }
+
+            @Override
+            public void deleteBranch(String repoRoot, String branch) {
+            }
+
+            @Override
+            public boolean hasUncommitted(String worktreePath) {
+                // Runs from INSIDE release()'s git-status shell-out: the registry entry is
+                // already gone, but the pane has not stopped yet.
+                duringWindow.set(resolverRef.get().resolve("127.0.0.1", 42, null));
+                return false;
+            }
+
+            @Override
+            public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+            }
+
+            @Override
+            public String repoRoot(String cwd) {
+                return "/repo";
+            }
+
+            @Override
+            public Optional<String> snapshot(String worktreePath, String branch, String message) {
+                return Optional.empty();
+            }
+
+            @Override
+            public WipRefStats wipRefs(String repoRoot) {
+                return new WipRefStats(0, 0L);
+            }
+
+            @Override
+            public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+                return 0;
+            }
+
+            @Override
+            public void shareWithGroup(String repoRoot, String worktreePath) {
+            }
+        };
+        SessionManager sessions = new SessionManager(workers, worktrees);
+
+        // The lead tab map names "term_a" before anything is ever spawned onto it — the control
+        // this test needs. Built up front so the SAME resolver answers every resolve() call below.
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> FakeHerdr.WORKER_PID);
+        CallerResolver resolver = CallerResolver.withLeadsAndMembers(identity, false, null,
+                () -> Map.of("term_a", "the-lead"), null, sessions::spawnedMemberRole, Map::of);
+        resolverRef.set(resolver);
+
+        Principal before = resolver.resolve("127.0.0.1", 42, null);
+        assertEquals(Role.PRIMARY, before.role(),
+                "control: with no live or releasing member on this terminal, the lead tab map must "
+                        + "win — this is what proves the in-window assertion below is not passing on "
+                        + "an empty map");
+        assertEquals("the-lead", before.name());
+
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702", null));
+        assertEquals("term_a", s.terminalId(), "sanity: the spawn resolved to the pinned pane");
+
+        sessions.release(s.paneId());
+
+        assertNotNull(duringWindow.get(), "the dirty check must have run and captured a resolve");
+        assertEquals(Role.WORKER, duringWindow.get().role(),
+                "inside the window the pane must resolve as its own live-member role, consulting no "
+                        + "tab map — a lead tab naming the same terminal must not win");
+        assertEquals("term_a", duringWindow.get().terminal());
+    }
+
+    /**
+     * {@link SessionManager#releaseRemoved} unbinds the architect slot before the git-status
+     * shell-out that opens the teardown window, so a resolve landing inside that window must see
+     * the slot already unbound and resolve {@link Role#WORKER} — never {@link Role#ARCHITECT},
+     * and never by checking role equality against the live session, which would hold even if the
+     * unbind ran too late.
+     *
+     * <p>The control is the same resolve taken outside the window, while the slot is still bound,
+     * which must return {@link Role#ARCHITECT} — without it this test would also pass against a
+     * slot that was never bound, and prove nothing.
+     */
+    @Test
+    void aReleasingArchitectIsDemotedToWorkerInsideTheTeardownWindow() {
+        FakeHerdr herdr = new FakeHerdr().pinNextStarts(1, "term_a", "w2:p7");
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
+                List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
+
+        AtomicReference<Principal> duringWindow = new AtomicReference<>();
+        AtomicReference<CallerResolver> resolverRef = new AtomicReference<>();
+        Worktrees worktrees = new Worktrees() {
+            @Override
+            public String add(String repoRoot, String branch, String baseRef) {
+                return "/wt/" + branch.replace('/', '_');
+            }
+
+            @Override
+            public void remove(String repoRoot, String worktreePath) {
+            }
+
+            @Override
+            public void deleteBranch(String repoRoot, String branch) {
+            }
+
+            @Override
+            public boolean hasUncommitted(String worktreePath) {
+                // Runs from INSIDE release()'s git-status shell-out: the architect slot is already
+                // unbound by this point, but the pane has not stopped yet.
+                duringWindow.set(resolverRef.get().resolve("127.0.0.1", 42, null));
+                return false;
+            }
+
+            @Override
+            public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+            }
+
+            @Override
+            public String repoRoot(String cwd) {
+                return "/repo";
+            }
+
+            @Override
+            public Optional<String> snapshot(String worktreePath, String branch, String message) {
+                return Optional.empty();
+            }
+
+            @Override
+            public WipRefStats wipRefs(String repoRoot) {
+                return new WipRefStats(0, 0L);
+            }
+
+            @Override
+            public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+                return 0;
+            }
+
+            @Override
+            public void shareWithGroup(String repoRoot, String worktreePath) {
+            }
+        };
+        SessionManager sessions = new SessionManager(workers, worktrees);
+
+        MemberRegistry members = new MemberRegistry(new FleetConfig.Fleet(Map.of(),
+                Map.of("lead-designer", new FleetConfig.Slot("ltms-local")), Map.of(), Map.of(), null));
+        sessions.setMemberLifecycle(members);
+
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> FakeHerdr.WORKER_PID);
+        CallerResolver resolver = CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, members, sessions::spawnedMemberRole, Map::of);
+        resolverRef.set(resolver);
+
+        MemberSession s = sessions.acquire("ltms-local", MemberRole.ARCHITECT, null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702d", null));
+        assertEquals("term_a", s.terminalId(), "sanity: the spawn resolved to the pinned pane");
+        assertEquals(MemberRole.ARCHITECT, s.role(), "sanity: the slot bind succeeded");
+
+        Principal before = resolver.resolve("127.0.0.1", 42, null);
+        assertEquals(Role.ARCHITECT, before.role(),
+                "control: with the slot still bound, the pane must resolve as an architect — this "
+                        + "is what proves the in-window assertion below is not passing against a "
+                        + "slot that was never bound");
+        assertEquals("lead-designer", before.name());
+
+        sessions.release(s.paneId());
+
+        assertNotNull(duringWindow.get(), "the dirty check must have run and captured a resolve");
+        assertEquals(Role.WORKER, duringWindow.get().role(),
+                "inside the window the architect slot is already unbound, so the result must be "
+                        + "WORKER — asserting role-equality with the live session here would tempt "
+                        + "moving the unbind earlier or later, which would be wrong either way");
+        assertEquals("term_a", duringWindow.get().terminal());
+    }
+
+    /** A spawned architect in the roster resolves ARCHITECT, carrying its bound slot's name. */
+    @Test
+    void aSpawnedArchitectInTheRosterResolvesArchitectWithItsSlotName() {
+        Principal p = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                        boundMembers("architect:lead-designer", MemberRole.ARCHITECT),
+                        t -> "term_a".equals(t) ? MemberRole.ARCHITECT : null, Map::of)
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.ARCHITECT, p.role());
+        assertEquals("lead-designer", p.name());
+        assertEquals("term_a", p.terminal());
+    }
+
+    /**
+     * fleetd #424 regression: the roster only answers THAT a pane is a live spawned member; config
+     * still decides WHAT that member's slot grants. A slot revoked after the bind must still demote
+     * the session on its very next request, exactly as it would for a pane with no roster entry at
+     * all — the roster's own ARCHITECT role must never be granted on its word alone.
+     *
+     * <p>{@code bind} refuses an unconfigured slot, so the revoked state can only be reached by
+     * binding while the slot is configured and then swapping the config out from under it, the way
+     * a live reload does.
+     */
+    @Test
+    void aRevokedArchitectSlotDemotesALiveSpawnedArchitectToWorker() {
+        FleetConfig.Fleet configured = new FleetConfig.Fleet(Map.of(),
+                Map.of("lead-designer", new FleetConfig.Slot("sonnet")), Map.of(), Map.of(), null);
+        java.util.concurrent.atomic.AtomicReference<FleetConfig.Fleet> live =
+                new java.util.concurrent.atomic.AtomicReference<>(configured);
+        MemberRegistry members = MemberRegistry.live(live::get);
+        assertTrue(members.bind("architect:lead-designer", "term_a"));
+
+        live.set(new FleetConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(), null)); // slot revoked
+
+        // Setup controls: the slot is really gone from config, but the occupancy is still there —
+        // otherwise this test would pass for the wrong reason.
+        assertNull(members.roleForSlot("architect:lead-designer"), "setup control: the slot must be gone from config");
+        assertEquals("architect:lead-designer", members.snapshot().get("term_a"),
+                "setup control: the binding itself must still be there");
+
+        Principal p = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                        members, t -> "term_a".equals(t) ? MemberRole.ARCHITECT : null, Map::of)
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.WORKER, p.role(),
+                "a revoked slot must demote a live spawned architect on its very next request");
+    }
+
+    /** Criterion 3: a configured collaborator tab that is not a spawned member resolves COLLABORATOR. */
+    @Test
+    void aConfiguredCollaboratorTabResolvesToCollaboratorCarryingItsName() {
+        Principal p = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                        new MemberRegistry(null), t -> null, () -> Map.of("term_a", "ops"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.COLLABORATOR, p.role());
+        assertEquals("ops", p.name());
+        assertEquals("term_a", p.terminal());
+    }
+
+    /** Regression: an empty collaborator registry leaves every pane exactly as before. */
+    @Test
+    void anEmptyCollaboratorRegistryLeavesEveryPaneAsBefore() {
+        Principal p = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                        new MemberRegistry(null), t -> null, Map::of)
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.WORKER, p.role());
+        assertNull(p.name());
+    }
+
+    @Test
+    void describeNamesTheCollaborator() {
+        assertEquals("collaborator:ops", Principal.collaborator("ops", "term_a", 1).describe());
+    }
+
+    // ── fleetd #669 Unit D: knownLeadOrCollaborator() reads the same maps resolve() does ───────────
+
+    @Test
+    void knownLeadOrCollaboratorIsTrueForALeadTerminal() {
+        CallerResolver r = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null,
+                () -> Map.of("term_lead", "opus-5.0"), new MemberRegistry(null), t -> null, Map::of);
+
+        assertTrue(r.knownLeadOrCollaborator().test("term_lead"));
+        assertFalse(r.knownLeadOrCollaborator().test("term_other"));
+    }
+
+    @Test
+    void knownLeadOrCollaboratorIsTrueForACollaboratorTerminal() {
+        CallerResolver r = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                new MemberRegistry(null), t -> null, () -> Map.of("term_collab", "ops"));
+
+        assertTrue(r.knownLeadOrCollaborator().test("term_collab"));
+        assertFalse(r.knownLeadOrCollaborator().test("term_other"));
+    }
+
+    @Test
+    void knownLeadOrCollaboratorIsFalseForASpawnedMembersTerminal() {
+        // The exact scenario a collaborator's SEND must never reach: a live spawned member's own
+        // terminal, which is neither a configured lead nor a configured collaborator.
+        CallerResolver r = CallerResolver.withLeadsAndMembers(workerIdentity(), false, null, Map::of,
+                new MemberRegistry(null), t -> "term_a".equals(t) ? MemberRole.DEV : null, Map::of);
+
+        assertFalse(r.knownLeadOrCollaborator().test("term_a"));
+    }
+
 }

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

@@ -107,6 +107,8 @@ class ConfigRefTest {
 
         assertTrue(out.applied());
         assertTrue(out.deferred().isEmpty());
+        assertTrue(out.split().stream().noneMatch(s -> s.contains("fleet.collaborators")),
+                out.split().toString());
         assertEquals("new charter", ref.get().fleet().charterFor(
                 dev.ltms.fleet.peer.MemberRole.ARCHITECT));
     }
@@ -786,14 +788,100 @@ class ConfigRefTest {
         assertTrue(out.split().getFirst().startsWith("fleet:"), out.split().toString());
         assertTrue(out.split().getFirst().contains("restart"), out.split().toString());
         assertTrue(out.split().getFirst().contains("live"), out.split().toString());
+        assertTrue(out.split().stream().noneMatch(s -> s.contains("fleet.collaborators")),
+                out.split().toString());
         assertTrue(out.summary().contains("partially live"), out.summary());
         // The snapshot still carries the new value — the LeadTabScanner's identity map and
         // LeadLauncher's auto-launch are what wait for a restart; a reload rebuilds neither.
         assertEquals("lead: opus-b", ref.get().fleet().leaders().get("opus").tab());
     }
 
+    @Test
+    void addingAFleetCollaboratorIsReportedAsSplit(@TempDir Path dir) throws Exception {
+        assertCollaboratorChangeIsReported(dir, """
+                fleet:
+                  collaborators:
+                    alex:
+                      tab: "collaborator: alex"
+                """, "add a collaborator");
+    }
+
+    @Test
+    void removingAFleetCollaboratorIsReportedAsSplit(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, yaml("""
+                fleet:
+                  collaborators:
+                    alex:
+                      tab: "collaborator: alex"
+                """));
+        ConfigRef ref = refFor(f);
+
+        Files.writeString(f, yaml("fleet: {}\n"));
+        assertCollaboratorSplit(ref.reload(), "remove a collaborator");
+    }
+
+    @Test
+    void changingAFleetCollaboratorTabIsReportedAsSplit(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, yaml("""
+                fleet:
+                  collaborators:
+                    alex:
+                      tab: "collaborator: alex-a"
+                """));
+        ConfigRef ref = refFor(f);
+
+        Files.writeString(f, yaml("""
+                fleet:
+                  collaborators:
+                    alex:
+                      tab: "collaborator: alex-b"
+                """));
+        assertCollaboratorSplit(ref.reload(), "change a collaborator tab");
+    }
+
+    @Test
+    void unchangedFleetCollaboratorsProduceNoSplitReport(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        String config = yaml("""
+                fleet:
+                  collaborators:
+                    alex:
+                      tab: "collaborator: alex"
+                """);
+        Files.writeString(f, config);
+        ConfigRef ref = refFor(f);
+
+        Files.writeString(f, config);
+        ConfigRef.Outcome out = ref.reload();
+
+        assertTrue(out.applied());
+        assertTrue(out.split().isEmpty(), out.split().toString());
+    }
+
+    private static void assertCollaboratorChangeIsReported(Path dir, String changed, String action)
+            throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, yaml("fleet: {}\n"));
+        ConfigRef ref = refFor(f);
+
+        Files.writeString(f, yaml(changed));
+        assertCollaboratorSplit(ref.reload(), action);
+    }
+
+    private static void assertCollaboratorSplit(ConfigRef.Outcome out, String action) {
+        assertTrue(out.applied(), action);
+        assertTrue(out.deferred().isEmpty(), out.deferred().toString());
+        assertEquals(1, out.split().size(), out.split().toString());
+        String report = out.split().getFirst();
+        assertTrue(report.startsWith("fleet: fleet.collaborators"), report);
+        assertTrue(report.contains("read once at startup"), report);
+        assertTrue(report.contains("restart"), report);
+    }
+
     /**
-     * fleetd #333: {@code fleet.leaders} is the ONLY frozen part of {@code fleet:}. A reload that
+     * fleetd #333: {@code fleet.leaders} is a frozen part of {@code fleet:}. A reload that
      * changes {@code tabLabel} (or charters, or a role pool) without touching {@code fleet.leaders}
      * must stay fully hot with nothing reported — proving {@link ConfigRef#changedSplitKeys}
      * compares {@code fleet.leaders} specifically rather than the whole {@code Fleet} record, which

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

@@ -850,7 +850,8 @@ class FleetConfigTest {
 
     /**
      * The hazard this guard closes: a pane-placed member lands inside the focused tab rather than
-     * its own, so it can land inside a lead's labelled tab and be read back as that lead.
+     * its own, so it can land inside a lead's labelled tab and, while its pane carries no entry in
+     * the spawned-member roster, be read back as that lead.
      */
     @Test
     void aPanePlacedProfileWithALeadTabRefusesToStart(@TempDir Path dir) throws Exception {
@@ -1087,8 +1088,9 @@ class FleetConfigTest {
     }
 
     /**
-     * fleetd #669: a member tabLabel that can render as a configured collaborator tab is the same
-     * hazard as the lead case above — a member labelled that way is read back as the collaborator.
+     * A member tabLabel that can render as a configured collaborator tab is the same hazard as the
+     * lead case above — while its pane carries no entry in the spawned-member roster, a member
+     * labelled that way is read back as the collaborator.
      */
     @Test
     void aProfileTabLabelOverrideMatchingACollaboratorTabRefusesToStart(@TempDir Path dir)

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

@@ -2,6 +2,8 @@ package dev.ltms.fleet.herdr;
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Map;
+
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
 
@@ -28,4 +30,44 @@ class HerdrRouterTest {
         assertSame(router.leadAgents(), router.agentsFor("lead"));
         assertSame(router.memberAgents(), router.agentsFor("member"));
     }
+
+    /**
+     * fleetd #669 Unit E. The predicate shape here is exactly what {@code FleetdAssembly} builds:
+     * true when the terminal is a known lead OR a known collaborator. Two distinct clients are
+     * required — with one shared client {@code agentsFor} would return the same object regardless
+     * of the predicate's answer, and this assertion would pass whether or not the collaborator map
+     * was ever consulted.
+     */
+    @Test
+    void collaboratorTerminalRoutesToTheLeadDaemon() {
+        FakeHerdr lead = new FakeHerdr();
+        FakeHerdr member = new FakeHerdr();
+        Map<String, String> leads = Map.of("term_lead", "primary");
+        Map<String, String> collaborators = Map.of("term_collab", "reviewer-alex");
+        HerdrRouter router = new HerdrRouter(lead, member,
+                id -> leads.containsKey(id) || collaborators.containsKey(id));
+
+        assertSame(router.leadAgents(), router.agentsFor("term_collab"),
+                "a configured collaborator's terminal must route to the LEAD daemon, not the member "
+                        + "one — its pane is opened by a person, exactly like a lead's");
+    }
+
+    /**
+     * Companion to {@link #collaboratorTerminalRoutesToTheLeadDaemon}: a terminal that is neither a
+     * known lead nor a known collaborator must still route to the member daemon. Without this, a
+     * predicate of {@code _ -> true} would also pass the test above.
+     */
+    @Test
+    void terminalInNeitherMapStillRoutesToTheMemberDaemon() {
+        FakeHerdr lead = new FakeHerdr();
+        FakeHerdr member = new FakeHerdr();
+        Map<String, String> leads = Map.of("term_lead", "primary");
+        Map<String, String> collaborators = Map.of("term_collab", "reviewer-alex");
+        HerdrRouter router = new HerdrRouter(lead, member,
+                id -> leads.containsKey(id) || collaborators.containsKey(id));
+
+        assertSame(router.memberAgents(), router.agentsFor("term_worker"),
+                "a terminal absent from both maps must stay on the member daemon — the fix widens "
+                        + "the predicate, it does not make it unconditionally true");
+    }
 }

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

@@ -163,6 +163,13 @@ class LeadTabScannerTest {
         return new LeadTabScanner(herdr, tabToName, Set.of("fleetd-workers"), TTL, clock::get);
     }
 
+    private LeadTabScanner scannerWithCollaborators(TopologyHerdr herdr, Map<String, String> tabToName,
+                                                     Map<String, String> collaboratorTabToName,
+                                                     AtomicLong clock) {
+        return new LeadTabScanner(herdr, tabToName, collaboratorTabToName,
+                Set.of("fleetd-workers"), TTL, clock::get);
+    }
+
     @Test
     void everyConfiguredTabBecomesALeadNamedByItsEntry() {
         Map<String, String> leads = scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong()).get();
@@ -198,9 +205,12 @@ class LeadTabScannerTest {
     }
 
     /**
-     * The guard that matters: fleetd labels its own worker tabs, so if a worker space were scanned
-     * a naming accident would promote the fleet. The exclusion is by workspace, not by hoping the
-     * worker template never collides.
+     * Covers the {@code excludedWorkspaceLabels} parameter: a tab in an excluded workspace is never
+     * matched, whatever its label. Production always constructs this class with an empty set (CB-558,
+     * {@code FleetdAssembly}), so this parameter plays no part in the live guard against a worker
+     * tab being mistaken for a lead — that guard is {@code CallerResolver} asking the live
+     * spawned-member roster before any tab map. This test exists because the parameter still exists
+     * and is worth covering on its own terms.
      */
     @Test
     void aTabInAWorkerSpaceIsNeverALeadEvenWhenItsLabelMatches() {
@@ -212,6 +222,29 @@ class LeadTabScannerTest {
         assertFalse(scanner(herdr, tabToName, new AtomicLong()).get().containsKey("term_impostor"));
     }
 
+    /**
+     * The shipped shape (CB-558, {@code FleetdAssembly}): production always constructs this class
+     * with an empty {@code excludedWorkspaceLabels}, and a lead's {@code workspace:} default is the
+     * same shared {@code "fleet"} space the members use. A lead tab is still found when it sits in
+     * the exact same workspace as a member-labelled tab — the scanner tells them apart by the exact
+     * tab label, not by which workspace either one is in.
+     */
+    @Test
+    void aLeadIsDiscoveredWhenItsWorkspaceIsTheSameAsTheMemberWorkspace() {
+        TopologyHerdr herdr = new TopologyHerdr()
+                .workspace("w1", "fleet")
+                .tab("w1:t1", "w1", "lead: opus-5.0")
+                .tab("w1:t2", "w1", "worker: gx10 #1")
+                .pane("w1:p1", "w1:t1", "term_opus")
+                .pane("w1:p2", "w1:t2", "term_worker");
+        LeadTabScanner s = new LeadTabScanner(herdr, Map.of("lead: opus-5.0", "opus-5.0"),
+                Set.of(), TTL, new AtomicLong()::get);
+
+        assertEquals("opus-5.0", s.get().get("term_opus"),
+                "a lead sharing the members' workspace is still discovered — the label, not the "
+                        + "workspace, is what matches it");
+    }
+
     @Test
     void aLabelWithNoConfiguredEntryIsIgnored() {
         TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
@@ -464,4 +497,75 @@ class LeadTabScannerTest {
 
         assertEquals(afterFirst, herdr.calls, "the failure path must be rate-limited too");
     }
+
+    // ── fleetd #669 Unit D: a collaborator tab is matched the same way as a lead tab, one pass ─────
+
+    @Test
+    void aConfiguredCollaboratorTabIsReportedByCollaboratorsNotByGet() {
+        TopologyHerdr herdr = new TopologyHerdr()
+                .workspace("w1", "main")
+                .tab("w1:t1", "w1", "collab: ops")
+                .pane("w1:p1", "w1:t1", "term_ops");
+        LeadTabScanner s = scannerWithCollaborators(herdr, Map.of(), Map.of("collab: ops", "ops"),
+                new AtomicLong());
+
+        assertEquals(Map.of("term_ops", "ops"), s.collaborators(),
+                "a collaborator tab is matched exactly like a lead tab");
+        assertEquals(Map.of(), s.get(), "a collaborator tab must never also appear as a lead");
+    }
+
+    /**
+     * Criterion 4, scanner level: a collaborator tab that is labelled but runs no agent is not
+     * reported — the same #359 liveness cross-check a lead tab gets.
+     */
+    @Test
+    void aDeadCollaboratorTabIsNotReported() {
+        TopologyHerdr herdr = new TopologyHerdr()
+                .workspace("w1", "main")
+                .tab("w1:t1", "w1", "collab: ops")
+                .pane("w1:p1", "w1:t1", "term_ops")
+                .deadAgent("w1:t1");
+        LeadTabScanner s = scannerWithCollaborators(herdr, Map.of(), Map.of("collab: ops", "ops"),
+                new AtomicLong());
+
+        assertFalse(s.collaborators().containsKey("term_ops"),
+                "a dead collaborator tab must never resolve as a live collaborator");
+    }
+
+    /**
+     * Both kinds are matched in a single pass over the same tab list — not a second scanner, not a
+     * second scan. Proven by herdr call count: scanning one lead tab and one collaborator tab in the
+     * same instance costs exactly as many calls as scanning two lead tabs in {@link #twoLeads()}.
+     */
+    @Test
+    void leadsAndCollaboratorsAreMatchedInOnePassOverTheSameScan() {
+        TopologyHerdr oneOfEach = new TopologyHerdr()
+                .workspace("w1", "main")
+                .tab("w1:t1", "w1", "lead: opus-5.0")
+                .tab("w1:t2", "w1", "collab: ops")
+                .pane("w1:p1", "w1:t1", "term_opus")
+                .pane("w1:p2", "w1:t2", "term_ops");
+        LeadTabScanner s = scannerWithCollaborators(oneOfEach, Map.of("lead: opus-5.0", "opus-5.0"),
+                Map.of("collab: ops", "ops"), new AtomicLong());
+
+        assertEquals(Map.of("term_opus", "opus-5.0"), s.get());
+        assertEquals(Map.of("term_ops", "ops"), s.collaborators());
+
+        TopologyHerdr twoLeadsBaseline = twoLeads();
+        scanner(twoLeadsBaseline, twoLeadsConfigured(), new AtomicLong()).get();
+
+        assertEquals(twoLeadsBaseline.calls, oneOfEach.calls,
+                "one lead tab + one collaborator tab must cost exactly as many herdr calls as two "
+                        + "lead tabs — proof this is one pass, not a second scan");
+    }
+
+    /** Regression: with no collaborators configured, every existing lead-only behaviour is unchanged. */
+    @Test
+    void anEmptyCollaboratorMapLeavesCollaboratorsEmptyAndGetUnaffected() {
+        LeadTabScanner s = scannerWithCollaborators(twoLeads(), twoLeadsConfigured(), Map.of(),
+                new AtomicLong());
+
+        assertEquals(Map.of(), s.collaborators());
+        assertEquals(Map.of("term_opus", "opus-5.0", "term_gpt", "gpt-sol-5.6"), s.get());
+    }
 }

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

@@ -63,6 +63,17 @@ class FleetMcpAuthzTest {
 
     /** A fully wired FleetMcp on fakes — constructing it is itself part of what is under test. */
     private FleetMcp mcp(boolean enforce) {
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> 999_999);
+        return mcp(enforce, CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, new MemberRegistry(null)));
+    }
+
+    /**
+     * As {@link #mcp(boolean)}, with an explicit {@link CallerResolver} — so a test can wire known
+     * leads/collaborators and drive {@code denyFor}'s real {@code knownLeadOrCollaborator()}
+     * classifier instead of the default empty one.
+     */
+    private FleetMcp mcp(boolean enforce, CallerResolver callers) {
         FleetConfig.Profile cfg = new FleetConfig.Profile(
                 "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
                 "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
@@ -83,9 +94,7 @@ class FleetMcpAuthzTest {
         // to be omitted to reach "legacy" is now always real, and AuthorizationMode is the
         // separate, explicit choice that governs enforcement.
         mcp = new FleetMcp(messages, workers, sessions, identity, sessions.asPresence(),
-                new PrimaryRegistry(null),
-                CallerResolver.withLeadsAndMembers(identity, false, null,
-                Map::of, new MemberRegistry(null)),
+                new PrimaryRegistry(null), callers,
                 enforce ? FleetMcp.AuthorizationMode.ENFORCED : FleetMcp.AuthorizationMode.UNENFORCED,
                 metrics, FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), null, FleetMcp.OutageSource.none(),
@@ -240,6 +249,26 @@ class FleetMcpAuthzTest {
         assertTrue(denied.isError());
     }
 
+    /**
+     * fleetd #669 Unit D: wires a real {@link CallerResolver} with a known lead and a known
+     * collaborator tab, and leaves a spawned member's own terminal recognised by neither map — so a
+     * collaborator's SEND reaches both named peers and is refused for the spawned member's terminal,
+     * over MCP's {@code denyFor}.
+     */
+    @Test
+    void aCollaboratorMaySendToAKnownLeadOrCollaboratorButNotToASpawnedMembersTerminal() {
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> 999_999);
+        CallerResolver callers = CallerResolver.withLeadsAndMembers(identity, false, null,
+                () -> Map.of("term_lead_known", "lead-x"), new MemberRegistry(null),
+                t -> null, () -> Map.of("term_collab_known", "ops2"));
+        FleetMcp m = mcp(true, callers);
+
+        assertNull(m.denyFor(COLLABORATOR, Authz.Action.SEND, "term_lead_known"));
+        assertNull(m.denyFor(COLLABORATOR, Authz.Action.SEND, "term_collab_known"));
+        assertNotNull(m.denyFor(COLLABORATOR, Authz.Action.SEND, "term_a"),
+                "a spawned member's own terminal must stay unreachable, even once the classifier is real");
+    }
+
     @Test
     void theLegacyConstructorLeavesTheGateOpen() {
         // The 22 pre-existing FleetMcpTest cases rely on no authorization being enforced.
@@ -326,10 +355,10 @@ class FleetMcpAuthzTest {
                         + "the anchors have drifted, this test is not testing what it claims to");
 
         Pattern trailingArg = Pattern.compile(
-                "listFleet\\([^;]*?,\\s*(coordinatorVisibleTo\\(principal\\(exchange\\)\\)|true|false)\\s*\\)\\s*;",
+                "listFleet\\([^;]*?,\\s*(coordinatorVisibleTo\\(principal\\(exchange\\)\\)|true|false)\\s*[,)]",
                 Pattern.DOTALL);
         Matcher m = trailingArg.matcher(handlerBlock);
-        assertTrue(m.find(), "could not locate listFleet(...)'s trailing boolean argument in the "
+        assertTrue(m.find(), "could not locate listFleet(...)'s coordinator boolean argument in the "
                 + "listHandler block -- the call shape changed, update this test's anchor: " + handlerBlock);
         String trailing = m.group(1);
         assertEquals("coordinatorVisibleTo(principal(exchange))", trailing,
@@ -337,6 +366,197 @@ class FleetMcpAuthzTest {
                         + "calling, not pass a literal boolean -- found: " + trailing);
     }
 
+    // --- fleetd #703: who may see fleet_list's collaborators array -------------------------------
+
+    /**
+     * fleetd #703: {@link FleetMcp#collaboratorsVisibleTo} is the whole policy decision for
+     * {@code fleet_list}'s {@code collaborators} array. Visible to exactly the roles that may
+     * {@code SEND} to a named peer -- the primary, an architect, and a collaborator itself -- never
+     * a worker, which holds {@code READ} but can never {@code SEND} to a collaborator, and never an
+     * anonymous caller.
+     */
+    @Test
+    void onlyPrimaryArchitectAndCollaboratorMaySeeTheCollaboratorsArray() {
+        assertTrue(FleetMcp.collaboratorsVisibleTo(PRIMARY), "the primary must see the collaborators array");
+        assertTrue(FleetMcp.collaboratorsVisibleTo(ARCH_DESIGN), "an architect must see the collaborators array");
+        assertTrue(FleetMcp.collaboratorsVisibleTo(COLLABORATOR), "a collaborator must see its own peer roster");
+        assertFalse(FleetMcp.collaboratorsVisibleTo(WORKER_A),
+                "a worker holds READ but can never SEND to a collaborator, so it must not see the array");
+        assertFalse(FleetMcp.collaboratorsVisibleTo(ANON), "authenticated as nothing must not see it either");
+    }
+
+    /**
+     * fleetd #703, same reasoning as {@link #theFleetListHandlerActuallyConsultsCoordinatorVisibleTo}:
+     * the predicate above can be perfectly correct while the one production call site never asks it.
+     * This reads {@code FleetMcp.java}'s own source and asserts the {@code fleet_list} handler's
+     * {@code listFleet(...)} call both threads {@code callers.collaborators()} into the payload and
+     * asks {@code collaboratorsVisibleTo(principal(exchange))} for the visibility flag, rather than a
+     * literal boolean or an empty map.
+     */
+    @Test
+    void theFleetListHandlerActuallyConsultsCollaboratorsVisibleTo() throws Exception {
+        String source = Files.readString(MCP_SOURCE);
+
+        int start = source.indexOf("listHandler =");
+        assertTrue(start >= 0, "could not find the fleet_list handler (listHandler) in " + MCP_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("stopHandler =", start);
+        assertTrue(end > start, "could not find the handler declared after listHandler to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does contain a call to listFleet(...) -- if this
+        // fails, the anchors above moved and the assertions below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("listFleet("),
+                "control failed: the scraped listHandler block contains no listFleet( call at all -- "
+                        + "the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("callers.collaborators()"),
+                "the fleet_list handler must thread callers.collaborators() into listFleet(...), not an "
+                        + "empty or literal map -- block: " + handlerBlock);
+        assertTrue(handlerBlock.contains("collaboratorsVisibleTo(principal(exchange))"),
+                "the fleet_list handler must ask collaboratorsVisibleTo(principal(exchange)) who is "
+                        + "calling, not pass a literal boolean -- block: " + handlerBlock);
+    }
+
+    // --- who may see fleet_list's leads and members arrays ---------------------------------------
+
+    /**
+     * {@link FleetMcp#leadsVisibleTo} is the whole policy decision for {@code fleet_list}'s
+     * {@code leads} array: visible to exactly the roles that may {@code SEND} to a lead -- the
+     * primary, an architect, and a collaborator -- never a worker, which holds {@code READ} but
+     * can never {@code SEND} at all, and never an anonymous caller.
+     */
+    @Test
+    void primaryArchitectAndCollaboratorMaySeeTheLeadsArray() {
+        assertTrue(FleetMcp.leadsVisibleTo(PRIMARY), "the primary must see the leads array");
+        assertTrue(FleetMcp.leadsVisibleTo(ARCH_DESIGN), "an architect must see the leads array");
+        assertTrue(FleetMcp.leadsVisibleTo(COLLABORATOR),
+                "a collaborator may SEND to a lead, so it must see the leads array to learn where");
+        assertFalse(FleetMcp.leadsVisibleTo(WORKER_A),
+                "a worker holds READ but can never SEND, so it must not see the leads array");
+        assertFalse(FleetMcp.leadsVisibleTo(ANON), "authenticated as nothing must not see it either");
+    }
+
+    /**
+     * As {@link #primaryArchitectAndCollaboratorMaySeeTheLeadsArray}, for the {@code members}
+     * array -- but a collaborator may {@code SEND} only to a lead or another collaborator, never
+     * to a spawned member, so it must not see this one.
+     */
+    @Test
+    void onlyPrimaryAndArchitectMaySeeTheMembersArray() {
+        assertTrue(FleetMcp.membersVisibleTo(PRIMARY), "the primary must see the members array");
+        assertTrue(FleetMcp.membersVisibleTo(ARCH_DESIGN), "an architect must see the members array");
+        assertFalse(FleetMcp.membersVisibleTo(WORKER_A),
+                "a worker holds READ but can never SEND, so it must not see the members array");
+        assertFalse(FleetMcp.membersVisibleTo(COLLABORATOR),
+                "a collaborator may SEND to a lead, never to a spawned member, so it must not see the members array");
+        assertFalse(FleetMcp.membersVisibleTo(ANON), "authenticated as nothing must not see it either");
+    }
+
+    /**
+     * Same reasoning as {@link #theFleetListHandlerActuallyConsultsCoordinatorVisibleTo}: the
+     * predicate above can be perfectly correct while the one production call site never asks it.
+     * This reads {@code FleetMcp.java}'s own source and asserts the {@code fleet_list} handler's
+     * {@code listFleet(...)} call asks {@code leadsVisibleTo(principal(exchange))} for the leads
+     * visibility flag, rather than a literal boolean.
+     */
+    @Test
+    void theFleetListHandlerActuallyConsultsLeadsVisibleTo() throws Exception {
+        String source = Files.readString(MCP_SOURCE);
+
+        int start = source.indexOf("listHandler =");
+        assertTrue(start >= 0, "could not find the fleet_list handler (listHandler) in " + MCP_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("stopHandler =", start);
+        assertTrue(end > start, "could not find the handler declared after listHandler to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does contain a call to listFleet(...) -- if this
+        // fails, the anchors above moved and the assertion below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("listFleet("),
+                "control failed: the scraped listHandler block contains no listFleet( call at all -- "
+                        + "the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("leadsVisibleTo(principal(exchange))"),
+                "the fleet_list handler must ask leadsVisibleTo(principal(exchange)) who is calling, "
+                        + "not pass a literal boolean -- block: " + handlerBlock);
+    }
+
+    /** As {@link #theFleetListHandlerActuallyConsultsLeadsVisibleTo}, for {@code membersVisibleTo}. */
+    @Test
+    void theFleetListHandlerActuallyConsultsMembersVisibleTo() throws Exception {
+        String source = Files.readString(MCP_SOURCE);
+
+        int start = source.indexOf("listHandler =");
+        assertTrue(start >= 0, "could not find the fleet_list handler (listHandler) in " + MCP_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("stopHandler =", start);
+        assertTrue(end > start, "could not find the handler declared after listHandler to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        assertTrue(handlerBlock.contains("listFleet("),
+                "control failed: the scraped listHandler block contains no listFleet( call at all -- "
+                        + "the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("membersVisibleTo(principal(exchange))"),
+                "the fleet_list handler must ask membersVisibleTo(principal(exchange)) who is calling, "
+                        + "not pass a literal boolean -- block: " + handlerBlock);
+    }
+
+    /**
+     * {@code fleet_poll{ticket}} must thread the calling connection's own terminal into
+     * {@link MessageService#poll(String, String)}, so a worker cannot read a ticket a different
+     * session created.
+     */
+    @Test
+    void theFleetPollHandlerActuallyThreadsCallerTerminalIntoPoll() throws Exception {
+        String source = Files.readString(MCP_SOURCE);
+
+        int start = source.indexOf("pollHandler =");
+        assertTrue(start >= 0, "could not find the fleet_poll handler (pollHandler) in " + MCP_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("ackHandler =", start);
+        assertTrue(end > start, "could not find the handler declared after pollHandler to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does call poll(...) -- if this fails, the anchors
+        // above moved and the assertions below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("poll(messages,"),
+                "control failed: the scraped pollHandler block contains no poll(messages, ...) call "
+                        + "at all -- the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("callerTerminal(exchange)"),
+                "the fleet_poll handler must thread callerTerminal(exchange) into poll(...), not omit "
+                        + "it or pass a literal null -- block: " + handlerBlock);
+    }
+
+    /**
+     * {@code fleet_status} must thread the calling connection's own terminal into
+     * {@link FleetMcp#status(MessageService, String, String)}, so a caller that did not create a
+     * worker's open delegation cannot read its pending question through the status handler either.
+     */
+    @Test
+    void theFleetStatusHandlerActuallyThreadsCallerTerminalIntoStatus() throws Exception {
+        String source = Files.readString(MCP_SOURCE);
+
+        int start = source.indexOf("statusHandler =");
+        assertTrue(start >= 0, "could not find the fleet_status handler (statusHandler) in " + MCP_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("pollHandler =", start);
+        assertTrue(end > start, "could not find the handler declared after statusHandler to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does call status(...) -- if this fails, the anchors
+        // above moved and the assertions below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("status(messages,"),
+                "control failed: the scraped statusHandler block contains no status(messages, ...) "
+                        + "call at all -- the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("callerTerminal(exchange)"),
+                "the fleet_status handler must thread callerTerminal(exchange) into status(...), not "
+                        + "omit it or pass a literal null -- block: " + handlerBlock);
+    }
+
     // --- which action each tool hands the gate (fleetd #272) ------------------------------------
 
     /**

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

@@ -121,6 +121,7 @@ class FleetMcpLeadContextGaugeWiringTest {
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.LoopHealthSource.none(), FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(),
                 FleetMcp.LeadSeatSource.none(), contextGauge, leadConfigDirs,
-                Map.of(LEAD_TERMINAL, LEAD_NAME), LEAD_TERMINAL, FleetMcp.CoordinationSource.none(), false);
+                Map.of(LEAD_TERMINAL, LEAD_NAME), LEAD_TERMINAL, Map.of(), false,
+                FleetMcp.CoordinationSource.none(), false, true, true);
     }
 }

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

@@ -10,6 +10,7 @@ import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.AgentStatus;
 import dev.ltms.fleet.herdr.FakeHerdr;
 import dev.ltms.fleet.inject.LoopWatchdog;
+import dev.ltms.fleet.lead.LeadContextGauge;
 import dev.ltms.fleet.herdr.PaneLocator;
 import dev.ltms.fleet.herdr.WorkspaceControl;
 import dev.ltms.fleet.inject.Injector;
@@ -691,7 +692,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true);
 
         String out = textOf(res);
         // fleetd #361: reports both which coord-id a peer must use to reach ME, and this daemon's
@@ -720,7 +721,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true);
 
         String out = textOf(res);
         assertTrue(out.contains("\"mailbox\":{\"status\":\"unknown\"}"), out);
@@ -779,19 +780,82 @@ class FleetMcpTest {
         SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
         FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
                 .withMailbox("mac-opus", LeadChannel.MailboxState.exists("mac-opus", 0, 1));
-        boolean callerIsPrimary = Principal.worker("term_a", 1).isPrimary();
+        Principal worker = Principal.worker("term_a", 1);
 
         McpSchema.CallToolResult res = FleetMcp.listFleet(
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), callerIsPrimary);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), worker.isPrimary(),
+                FleetMcp.leadsVisibleTo(worker), FleetMcp.membersVisibleTo(worker));
 
         String out = textOf(res);
         assertFalse(out.contains("\"coordinator\""), "a worker must never see the coordinator key at all: " + out);
         assertFalse(out.contains("mac-opus"), "no fragment of the coordinator row may leak either: " + out);
-        assertTrue(out.contains("\"leads\""), "the rest of the result must still be present: " + out);
-        assertTrue(out.contains("\"members\""), out);
+        assertFalse(out.contains("\"leads\""), "a worker must never see the leads key at all: " + out);
+        assertFalse(out.contains("\"members\""), "a worker must never see the members key at all: " + out);
+        assertTrue(out.contains("\"healthCoverage\""), "the rest of the result must still be present: " + out);
+    }
+
+    /**
+     * A worker's result must carry neither the {@code leads} nor the {@code members} key, and no
+     * fragment of either row leaks even though both are fully populated for this call -- a key
+     * check alone would pass on an implementation that still built the rows and only renamed or
+     * nested them.
+     */
+    @Test
+    void listLeaksNoLeadOrMemberRowFragmentToAWorker() {
+        FakeHerdr h = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt");
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", "term_owner",
+                new WorktreeRequest("cb-304", null));
+        Principal worker = Principal.worker("term_a", 1);
+
+        McpSchema.CallToolResult res = FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
+                Map.of("term_lead_x", "mac-opus"), "", FleetMcp.CoordinationSource.none(), worker.isPrimary(),
+                FleetMcp.leadsVisibleTo(worker), FleetMcp.membersVisibleTo(worker));
+
+        String out = textOf(res);
+        assertFalse(out.contains("\"leads\""), out);
+        assertFalse(out.contains("\"members\""), out);
+        assertFalse(out.contains(s.terminalId()), "no member row fragment may leak to a worker: " + out);
+        assertFalse(out.contains("term_owner"), "no member owner fragment may leak to a worker: " + out);
+        assertFalse(out.contains("mac-opus"), "no lead row fragment may leak to a worker: " + out);
+        assertTrue(out.contains("\"healthCoverage\""), out);
+    }
+
+    /**
+     * A collaborator may {@code SEND} to a lead, so it must see the {@code leads} array -- the
+     * only place {@code fleet_whoami} does not already give it a lead's address. It may never
+     * {@code SEND} to a spawned member, so the {@code members} key must stay absent for it, with
+     * no fragment of a populated member row leaking either.
+     */
+    @Test
+    void listShowsLeadsButNotMembersToACollaborator() {
+        FakeHerdr h = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt");
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", "term_owner",
+                new WorktreeRequest("cb-304", null));
+        Principal collaborator = Principal.collaborator("ops", "term_collab", 600);
+
+        McpSchema.CallToolResult res = FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
+                Map.of("term_lead_x", "mac-opus"), "", FleetMcp.CoordinationSource.none(), collaborator.isPrimary(),
+                FleetMcp.leadsVisibleTo(collaborator), FleetMcp.membersVisibleTo(collaborator));
+
+        String out = textOf(res);
+        assertTrue(out.contains("\"leads\""), "a collaborator must see the leads array: " + out);
+        assertTrue(out.contains("mac-opus"), "a collaborator must see the lead's name/address: " + out);
+        assertFalse(out.contains("\"members\""), "a collaborator must never see the members key: " + out);
+        assertFalse(out.contains(s.terminalId()), "no member row fragment may leak to a collaborator: " + out);
+        assertFalse(out.contains("term_owner"), "no member owner fragment may leak to a collaborator: " + out);
         assertTrue(out.contains("\"healthCoverage\""), out);
     }
 
@@ -807,16 +871,19 @@ class FleetMcpTest {
         SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
         FakeLeadChannel channel = new FakeLeadChannel("mac-opus")
                 .withMailbox("mac-opus", LeadChannel.MailboxState.exists("mac-opus", 0, 1));
-        boolean callerIsPrimary = Principal.architect("lead-designer", "term_design", 400).isPrimary();
+        Principal architect = Principal.architect("lead-designer", "term_design", 400);
 
         McpSchema.CallToolResult res = FleetMcp.listFleet(
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), callerIsPrimary);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), architect.isPrimary(),
+                FleetMcp.leadsVisibleTo(architect), FleetMcp.membersVisibleTo(architect));
 
         String out = textOf(res);
         assertFalse(out.contains("\"coordinator\""), "an architect must never see the coordinator key either: " + out);
+        assertTrue(out.contains("\"leads\""), "an architect must still see the leads array: " + out);
+        assertTrue(out.contains("\"members\""), "an architect must still see the members array: " + out);
     }
 
     /**
@@ -841,7 +908,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true));
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true));
 
         assertTrue(gatedAsPrimary.contains("\"coordinator\""), gatedAsPrimary);
         assertTrue(gatedAsPrimary.contains("\"selfId\":\"mac-opus\""), gatedAsPrimary);
@@ -850,6 +917,8 @@ class FleetMcpTest {
         assertTrue(gatedAsPrimary.contains("\"heldDurable\""), gatedAsPrimary);
         assertTrue(gatedAsPrimary.contains("\"held\""), gatedAsPrimary);
         assertTrue(gatedAsPrimary.contains("\"peers\""), gatedAsPrimary);
+        assertTrue(gatedAsPrimary.contains("\"leads\""), "the primary must still see the leads array: " + gatedAsPrimary);
+        assertTrue(gatedAsPrimary.contains("\"members\""), "the primary must still see the members array: " + gatedAsPrimary);
     }
 
     @Test
@@ -868,7 +937,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true);
 
         String out = textOf(res);
         assertTrue(out.contains("\"msgId\":\"m1\""), out);
@@ -879,6 +948,83 @@ class FleetMcpTest {
         assertTrue(out.contains("x".repeat(80) + "…"), "expected an 80-char preview with an ellipsis: " + out);
     }
 
+    // --- fleetd #703: fleet_list's collaborators array -------------------------------------------
+
+    /** Calls the canonical {@code listFleet} overload directly, so a test can set the collaborators
+     * payload and its visibility independently of a real {@code Principal} / MCP exchange. */
+    private static McpSchema.CallToolResult listFleetWithCollaborators(FakeHerdr h,
+            Map<String, String> collaborators, boolean collaboratorsVisible) {
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        return FleetMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
+                FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
+                FleetMcp.LoopHealthSource.none(), FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(),
+                FleetMcp.LeadSeatSource.none(), new LeadContextGauge(), FleetMcp.LeadConfigDirSource.none(),
+                Map.of(), "", collaborators, collaboratorsVisible,
+                FleetMcp.CoordinationSource.none(), false, true, true);
+    }
+
+    /**
+     * fleetd #703 acceptance A: a visible caller with one configured collaborator gets a
+     * {@code collaborators} row whose key ({@code CallerResolver.collaborators()}'s
+     * {@code terminal_id -> name} entry) lands as that row's {@code sessionId}, and {@code leads}/
+     * {@code members} are unaffected by the new key.
+     */
+    @Test
+    void listReportsACollaboratorsRowKeyedByTheCollaboratorsTerminalId() {
+        FakeHerdr h = new FakeHerdr();
+        String out = textOf(listFleetWithCollaborators(h, Map.of("term_collab", "ops"), true));
+
+        assertTrue(out.contains("\"collaborators\":["), out);
+        assertTrue(out.contains("\"name\":\"ops\""), out);
+        assertTrue(out.contains("\"sessionId\":\"term_collab\""), out);
+        assertTrue(out.contains("\"leads\":[]"), out);
+        assertTrue(out.contains("\"members\":[]"), out);
+    }
+
+    /**
+     * fleetd #703 acceptance A, control half: with no collaborator configured, the key is absent
+     * (caller cannot see it) or an empty array (caller can), and {@code leads}/{@code members} are
+     * unchanged either way.
+     */
+    @Test
+    void listOmitsOrEmptiesCollaboratorsWhenNoneAreConfigured() {
+        FakeHerdr h = new FakeHerdr();
+
+        String visibleButEmpty = textOf(listFleetWithCollaborators(h, Map.of(), true));
+        assertTrue(visibleButEmpty.contains("\"collaborators\":[]"), visibleButEmpty);
+        assertTrue(visibleButEmpty.contains("\"leads\":[]"), visibleButEmpty);
+        assertTrue(visibleButEmpty.contains("\"members\":[]"), visibleButEmpty);
+
+        String notVisible = textOf(listFleetWithCollaborators(h, Map.of(), false));
+        assertFalse(notVisible.contains("\"collaborators\""), notVisible);
+        assertTrue(notVisible.contains("\"leads\":[]"), notVisible);
+        assertTrue(notVisible.contains("\"members\":[]"), notVisible);
+    }
+
+    /**
+     * fleetd #703 acceptance B: a worker must not see the {@code collaborators} array at all, while
+     * an architect -- a role that also holds READ, same as a worker -- does see it. Both halves are
+     * asserted: a test that only checked the worker-hidden half would pass even if the feature were
+     * never wired up for anyone.
+     */
+    @Test
+    void listOmitsCollaboratorsForAWorkerAndIncludesThemForAnArchitect() {
+        FakeHerdr h = new FakeHerdr();
+        Map<String, String> collaborators = Map.of("term_collab", "ops");
+
+        String asWorker = textOf(listFleetWithCollaborators(h, collaborators,
+                FleetMcp.collaboratorsVisibleTo(Principal.worker("term_w", 1))));
+        assertFalse(asWorker.contains("\"collaborators\""),
+                "a worker must not see the collaborators array: " + asWorker);
+
+        String asArchitect = textOf(listFleetWithCollaborators(h, collaborators,
+                FleetMcp.collaboratorsVisibleTo(Principal.architect("design", "term_arch", 2))));
+        assertTrue(asArchitect.contains("\"collaborators\":["),
+                "an architect must see the collaborators array: " + asArchitect);
+        assertTrue(asArchitect.contains("\"sessionId\":\"term_collab\""), asArchitect);
+    }
+
     /**
      * fleetd #421: {@code mailbox.pending} counts only broker-ready messages, so a blocked lead's
      * normal, healthy state is {@code "pending": 0} next to a non-empty {@code held[]} — which
@@ -900,7 +1046,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true);
 
         String out = textOf(res);
         assertTrue(out.contains("\"pending\":0"), out);
@@ -929,7 +1075,7 @@ class FleetMcpTest {
                 workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
-                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true);
+                Map.of(), "", new FleetMcp.CoordinationSource(channel, List.of()), true, true, true);
 
         String out = textOf(res);
         assertTrue(out.contains("\"heldDurable\":false"),
@@ -998,7 +1144,7 @@ class FleetMcpTest {
                 FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
                 FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none(), FleetMcp.LeadSeatSource.none(),
                 Map.of(), "",
-                new FleetMcp.CoordinationSource(channel, List.of("fleet01-lead", "fleet02-lead", "fleet03-lead")), true);
+                new FleetMcp.CoordinationSource(channel, List.of("fleet01-lead", "fleet02-lead", "fleet03-lead")), true, true, true);
 
         String out = textOf(res);
         assertTrue(out.contains("\"coordId\":\"fleet01-lead\",\"status\":\"exists\",\"pending\":2,\"consumers\":1"), out);
@@ -1714,14 +1860,14 @@ class FleetMcpTest {
         FakeHerdr blocked = new FakeHerdr().agentStatus("blocked");
         AgentControl blockedAgents = new AgentControl(blocked);
         McpSchema.CallToolResult res = FleetMcp.status(
-                new MessageService(blockedAgents, new Injector(blockedAgents), rendezvous), "term_a");
+                new MessageService(blockedAgents, new Injector(blockedAgents), rendezvous), "term_a", null);
         assertNotEquals(Boolean.TRUE, res.isError());
         assertEquals("blocked", textOf(res));
     }
 
     /**
-     * CB-582: a lead polling {@code fleet_status} on its normal cadence — not {@code fleet_poll}
-     * — must also see a worker's open async {@code fleet_ask} question, since the reverse-rendezvous
+     * A lead polling {@code fleet_status} on its normal cadence — not {@code fleet_poll} — must
+     * also see a worker's open async {@code fleet_ask} question, since the reverse-rendezvous
      * window it opened with is far shorter than that cadence.
      */
     @Test
@@ -1744,7 +1890,7 @@ class FleetMcpTest {
         } while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline);
         assertEquals(MessageService.Phase.ASKING, asking.phase());
 
-        McpSchema.CallToolResult res = FleetMcp.status(messages, T);
+        McpSchema.CallToolResult res = FleetMcp.status(messages, T, null);
         assertNotEquals(Boolean.TRUE, res.isError());
         String out = textOf(res);
         assertTrue(out.startsWith("idle"), "the live status must still lead the text: " + out);
@@ -1765,6 +1911,63 @@ class FleetMcpTest {
         answer.get(5, TimeUnit.SECONDS);
     }
 
+    /**
+     * {@code fleet_status}'s pending-ask block (the question, its {@code turnId} and its ticket)
+     * is shown only to the caller whose terminal created the delegation, or to a caller with no
+     * terminal at all (the unnamed primary) — a different terminal-bearing caller still sees the
+     * base status line, but none of the pending-ask fields.
+     */
+    @Test
+    void statusGatesThePendingAskFieldsByTheDelegationsCreatorTerminal() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks", null, "term_creator");
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting(T), "sendAsync should have opened its rendezvous waiter");
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+
+        MessageService.TaskView asking;
+        deadline = System.currentTimeMillis() + 3000;
+        do {
+            asking = messages.poll(ticket);
+            Thread.sleep(5);
+        } while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline);
+        assertEquals(MessageService.Phase.ASKING, asking.phase());
+
+        String other = textOf(FleetMcp.status(messages, T, "term_other"));
+        assertTrue(other.startsWith("idle"), "the base status must still be shown: " + other);
+        assertFalse(other.contains("which config file?"),
+                "a non-creating caller must not see the question text: " + other);
+        assertFalse(other.contains(asking.turnId()),
+                "a non-creating caller must not see the turnId: " + other);
+        assertFalse(other.contains(ticket),
+                "a non-creating caller must not see the ticket: " + other);
+
+        String creator = textOf(FleetMcp.status(messages, T, "term_creator"));
+        assertTrue(creator.contains("which config file?"), "the creator must see the question: " + creator);
+        assertTrue(creator.contains(asking.turnId()), "the creator must see the turnId: " + creator);
+        assertTrue(creator.contains(ticket), "the creator must see the ticket: " + creator);
+
+        String unnamed = textOf(FleetMcp.status(messages, T, null));
+        assertTrue(unnamed.contains("which config file?"),
+                "a caller with no terminal (the unnamed primary) must see the question: " + unnamed);
+
+        // Clean up the still-open ask so the background thread does not linger past the test.
+        String turnId = asking.turnId();
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(turnId, "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.resolve(T, "done"));
+        answer.get(5, TimeUnit.SECONDS);
+    }
+
     // --- fleet_whoami: the caller's own identity, so an agent never has to guess its role -------
 
     @Test

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

@@ -0,0 +1,280 @@
+package dev.ltms.fleet.msg;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Pins that no file under {@code src/main/java} calls the fail-open
+ * {@link MessageService#poll(String)} overload. That overload skips the ownership check in
+ * {@code MessageService}'s {@code ownsTicket} entirely, so a caller of it can read any session's
+ * ticket. Every production caller must go through {@link MessageService#poll(String, String)}
+ * and pass a {@code callerTerminal} explicitly, even when it is {@code null}.
+ *
+ * <p>This reads each file's own source text rather than reflecting on compiled bytecode, because
+ * the risk is a future one-word edit at a call site, not a missing overload.
+ *
+ * <p>The scan below finds a violation by its receiver, {@code messages.poll(}, rather than the
+ * bare method name, so it does not mistake {@link java.util.Queue#poll()} for a violation. That
+ * anchor only covers a {@code MessageService} reached through a variable or field named
+ * {@code messages}, so {@link #everyMessageServiceDeclarationIsNamedMessages} pins the naming
+ * convention the anchor depends on: a declaration under any other name would be invisible to the
+ * scan above, and must turn this second check red instead of passing silently.
+ */
+class MessageServicePollUsageTest {
+
+    private static final Path PRODUCTION_SOURCE = Path.of("src/main/java");
+
+    @Test
+    void noProductionFileCallsTheSingleArgumentPollOverload() throws IOException {
+        List<String> violations = new ArrayList<>();
+        List<String> twoArgSites = new ArrayList<>();
+        int filesScanned = scanForPollCalls(PRODUCTION_SOURCE, violations, twoArgSites);
+
+        // CONTROL: the scan actually walked files -- a wrong root would otherwise report "no
+        // violations found" having looked at nothing.
+        assertTrue(filesScanned > 0, "control failed: the scan under " + PRODUCTION_SOURCE
+                + " visited zero .java files -- the path is wrong, so the absence of violations "
+                + "below proves nothing");
+
+        assertTrue(violations.isEmpty(), "found a call to the fail-open MessageService.poll(String) "
+                + "overload, which skips the ownership check entirely -- pass a callerTerminal "
+                + "explicitly (even if null) through poll(String, String) instead: " + violations);
+
+        // CONTROL: the arity parser actually finds the two genuine two-argument call sites (the
+        // MCP handler in FleetMcp and the REST handler in FleetApp). If this drops, the parser
+        // itself is broken, not the production code -- a broken parser (or a scan root that
+        // reaches no real source) must fail loudly here rather than pass vacuously above.
+        assertEquals(2, twoArgSites.size(), "control failed: expected exactly the two known "
+                + "two-argument messages.poll(...) call sites, found: " + twoArgSites);
+        assertTrue(twoArgSites.stream().anyMatch(s -> s.contains("FleetMcp.java")),
+                "control failed: did not find the FleetMcp.java messages.poll(ticket, callerTerminal) "
+                        + "site among: " + twoArgSites);
+        assertTrue(twoArgSites.stream().anyMatch(s -> s.contains("FleetApp.java")),
+                "control failed: did not find the FleetApp.java messages.poll(...) site among: "
+                        + twoArgSites);
+    }
+
+    /**
+     * The {@code messages.poll(} anchor above only sees a {@code MessageService} reached through
+     * a variable, field, or parameter named {@code messages}. This asserts that every such
+     * declaration under {@code src/main/java} uses that name, so a differently named declaration
+     * -- invisible to the scan above -- fails loudly here instead of letting that scan pass on a
+     * call site it never looked at.
+     */
+    @Test
+    void everyMessageServiceDeclarationIsNamedMessages() throws IOException {
+        List<String> names = new ArrayList<>();
+        int filesScanned = scanForDeclarationNames(PRODUCTION_SOURCE, names);
+
+        // CONTROL: the scan actually walked files -- a wrong root would otherwise report "every
+        // declaration is named messages" having looked at nothing.
+        assertTrue(filesScanned > 0, "control failed: the scan under " + PRODUCTION_SOURCE
+                + " visited zero .java files -- the path is wrong, so the result below proves nothing");
+
+        // CONTROL: the declaration pattern actually finds real declarations. Zero means the
+        // pattern is broken, not that every MessageService variable, field, or parameter vanished.
+        assertTrue(names.size() > 0, "control failed: found zero MessageService declarations under "
+                + PRODUCTION_SOURCE + " -- the declaration pattern is broken, update it before "
+                + "trusting the naming check below");
+
+        List<String> other = names.stream().filter(n -> !n.equals("messages")).distinct().toList();
+        assertTrue(other.isEmpty(), "found a MessageService declaration not named \"messages\": "
+                + other + " -- the messages.poll( scan above only looks for that name, so a call "
+                + "through a differently named variable or field is invisible to it; either rename "
+                + "the declaration or widen that scan's anchor to cover it");
+    }
+
+    private static int scanForPollCalls(Path root, List<String> violations, List<String> twoArgSites)
+            throws IOException {
+        List<Path> files = javaFiles(root);
+        for (Path file : files) {
+            scanFileForPollCalls(file, violations, twoArgSites);
+        }
+        return files.size();
+    }
+
+    private static int scanForDeclarationNames(Path root, List<String> names) throws IOException {
+        List<Path> files = javaFiles(root);
+        Pattern declaration = Pattern.compile("MessageService\\s+([A-Za-z_][A-Za-z0-9_]*)");
+        for (Path file : files) {
+            if (file.getFileName().toString().equals("MessageService.java")) {
+                continue; // the type's own declaration, not a caller holding a reference to it
+            }
+            String stripped = stripComments(Files.readString(file));
+            Matcher m = declaration.matcher(stripped);
+            while (m.find()) {
+                int j = m.end();
+                while (j < stripped.length() && Character.isWhitespace(stripped.charAt(j))) j++;
+                if (j < stripped.length() && stripped.charAt(j) == '(') {
+                    continue; // a method named like the convention, e.g. "MessageService messages()"
+                }
+                names.add(m.group(1));
+            }
+        }
+        return files.size();
+    }
+
+    private static List<Path> javaFiles(Path root) throws IOException {
+        try (Stream<Path> paths = Files.walk(root)) {
+            return paths.filter(p -> p.toString().endsWith(".java")).toList();
+        }
+    }
+
+    /**
+     * Replaces {@code //} and {@code /* *}{@code /} comment text with nothing, leaving code,
+     * string/char literals and line breaks untouched -- so a comment that merely mentions
+     * {@code MessageService} in prose can never be read as a declaration.
+     */
+    private static String stripComments(String source) {
+        StringBuilder out = new StringBuilder(source.length());
+        boolean inString = false;
+        boolean inChar = false;
+        int i = 0;
+        while (i < source.length()) {
+            char c = source.charAt(i);
+            if (inString) {
+                out.append(c);
+                if (c == '\\' && i + 1 < source.length()) { out.append(source.charAt(i + 1)); i += 2; continue; }
+                if (c == '"') inString = false;
+                i++;
+                continue;
+            }
+            if (inChar) {
+                out.append(c);
+                if (c == '\\' && i + 1 < source.length()) { out.append(source.charAt(i + 1)); i += 2; continue; }
+                if (c == '\'') inChar = false;
+                i++;
+                continue;
+            }
+            if (c == '"') { inString = true; out.append(c); i++; continue; }
+            if (c == '\'') { inChar = true; out.append(c); i++; continue; }
+            if (c == '/' && i + 1 < source.length() && source.charAt(i + 1) == '/') {
+                while (i < source.length() && source.charAt(i) != '\n') i++;
+                continue; // leaves the newline itself for the next iteration to append
+            }
+            if (c == '/' && i + 1 < source.length() && source.charAt(i + 1) == '*') {
+                i += 2;
+                while (i < source.length() && !(source.charAt(i) == '*' && i + 1 < source.length()
+                        && source.charAt(i + 1) == '/')) {
+                    if (source.charAt(i) == '\n') out.append('\n');
+                    i++;
+                }
+                i += 2;
+                continue;
+            }
+            out.append(c);
+            i++;
+        }
+        return out.toString();
+    }
+
+    private static void scanFileForPollCalls(Path file, List<String> violations, List<String> twoArgSites)
+            throws IOException {
+        String source = Files.readString(file);
+        String needle = "messages.poll(";
+        int from = 0;
+        int idx;
+        while ((idx = source.indexOf(needle, from)) >= 0) {
+            int argsStart = idx + needle.length();
+            String args = extractBalancedArgs(source, argsStart, file, idx);
+            int closeParenIndex = argsStart + args.length();
+            from = closeParenIndex + 1;
+
+            if (args.isBlank()) {
+                continue; // MessageService has no zero-argument poll() -- nothing to classify
+            }
+            String site = file + ":" + lineOf(source, idx);
+            if (topLevelCommaCount(args) == 0) {
+                violations.add(site + " -- messages.poll(" + args.trim() + ")");
+            } else {
+                twoArgSites.add(site);
+            }
+        }
+    }
+
+    /**
+     * The text between {@code messages.poll(} and its matching close paren: balanced over nested
+     * calls, and never split by a paren or comma sitting inside a string or char literal.
+     */
+    private static String extractBalancedArgs(String source, int start, Path file, int callIndex) {
+        int depth = 1;
+        boolean inString = false;
+        boolean inChar = false;
+        int i = start;
+        while (i < source.length()) {
+            char c = source.charAt(i);
+            if (inString) {
+                if (c == '\\') { i += 2; continue; }
+                if (c == '"') inString = false;
+            } else if (inChar) {
+                if (c == '\\') { i += 2; continue; }
+                if (c == '\'') inChar = false;
+            } else if (c == '"') {
+                inString = true;
+            } else if (c == '\'') {
+                inChar = true;
+            } else if (c == '(') {
+                depth++;
+            } else if (c == ')') {
+                depth--;
+                if (depth == 0) return source.substring(start, i);
+            }
+            i++;
+        }
+        throw new IllegalStateException(
+                "unbalanced parentheses scanning " + file + ":" + lineOf(source, callIndex));
+    }
+
+    /**
+     * Commas at paren/bracket/brace depth zero, skipping string and char literals -- the argument
+     * separators a human reader would see, not every comma character in the text.
+     */
+    private static int topLevelCommaCount(String args) {
+        int depth = 0;
+        int commas = 0;
+        boolean inString = false;
+        boolean inChar = false;
+        int i = 0;
+        while (i < args.length()) {
+            char c = args.charAt(i);
+            if (inString) {
+                if (c == '\\') { i += 2; continue; }
+                if (c == '"') inString = false;
+            } else if (inChar) {
+                if (c == '\\') { i += 2; continue; }
+                if (c == '\'') inChar = false;
+            } else if (c == '"') {
+                inString = true;
+            } else if (c == '\'') {
+                inChar = true;
+            } else if (c == '(' || c == '[' || c == '{') {
+                depth++;
+            } else if (c == ')' || c == ']' || c == '}') {
+                depth--;
+            } else if (c == ',' && depth == 0) {
+                commas++;
+            }
+            i++;
+        }
+        return commas;
+    }
+
+    private static int lineOf(String source, int index) {
+        int line = 1;
+        for (int i = 0; i < index; i++) {
+            if (source.charAt(i) == '\n') line++;
+        }
+        return line;
+    }
+}

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

@@ -865,11 +865,82 @@ class MessageServiceTest {
                         + "follow-up send would come back BUSY instead of timing out on its own work");
     }
 
+    /**
+     * Drive an async send on {@code T} to a resolved reply, polling as {@code owner} until the
+     * ticket reports {@link MessageService.Phase#DONE} (or the 2s deadline runs out). {@code owner}
+     * must be a terminal this ticket's creator check actually accepts, or this loops to the
+     * deadline and returns a non-{@code DONE} view.
+     */
+    private MessageService.TaskView driveAsyncTicketToDone(String ticket, String owner) throws InterruptedException {
+        MessageService.TaskView view = null;
+        long deadline = System.currentTimeMillis() + 2000;
+        while (view == null || view.phase() != MessageService.Phase.DONE) {
+            if (System.currentTimeMillis() >= deadline) break;
+            view = messages.poll(ticket, owner);
+            //noinspection BusyWait
+            Thread.sleep(5);
+        }
+        return view;
+    }
+
     @Test
     void pollReturnsNullForAnUnknownTicket() {
         assertNull(messages.poll("task-999999"), "a ticket that was never minted is unknown");
     }
 
+    // --- fleetd #705: a ticket's creator terminal gates who may poll it -------------------------
+
+    @Test
+    void pollByAnotherTerminalIsRefused() throws Exception {
+        String ticket = messages.sendAsync(T, "long task", null, "term_a");
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker works
+        assertTrue(rendezvous.resolve(T, "secret async result"), "a reply resolves the async send");
+
+        MessageService.TaskView owner = driveAsyncTicketToDone(ticket, "term_a");
+        assertNotNull(owner, "the creator must still be able to read its own ticket");
+        assertEquals(MessageService.Phase.DONE, owner.phase());
+
+        MessageService.TaskView refused = messages.poll(ticket, "term_b");
+        assertNotNull(refused, "a different terminal gets a refusal, not silence");
+        assertNotEquals(MessageService.Phase.DONE, refused.phase(),
+                "a different terminal must never see the ticket as DONE");
+        assertNull(refused.reply(), "a refusal must never carry the reply text");
+        assertFalse(String.valueOf(refused).contains("secret async result"),
+                "the reply text must not appear anywhere in the refused view");
+    }
+
+    @Test
+    void unnamedPrimaryStillReadsAnyTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "long task", null, "term_lead");
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker works
+        assertTrue(rendezvous.resolve(T, "primary-visible result"), "a reply resolves the async send");
+
+        // callerTerminal == null is the unnamed primary (resolved by token or loopback trust, with
+        // no herdr pane) — it must read a ticket a terminal-bearing lead created.
+        MessageService.TaskView view = driveAsyncTicketToDone(ticket, null);
+        assertNotNull(view, "the unnamed primary must be able to read any ticket");
+        assertEquals(MessageService.Phase.DONE, view.phase());
+        assertEquals("primary-visible result", view.reply());
+    }
+
+    @Test
+    void creatorReadsItsOwnTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "long task", null, "term_creator");
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker works
+        assertTrue(rendezvous.resolve(T, "own result"), "a reply resolves the async send");
+
+        MessageService.TaskView view = driveAsyncTicketToDone(ticket, "term_creator");
+        assertNotNull(view, "the ticket's own creator must be able to read it");
+        assertEquals(MessageService.Phase.DONE, view.phase());
+        assertEquals("own result", view.reply());
+    }
+
     @Test
     void pollReportsACompletedTicket() throws Exception {
         String ticket = messages.sendAsync(T, "long task");
@@ -1804,20 +1875,20 @@ class MessageServiceTest {
         }
     }
 
-    // --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
+    // --- fleet_status pendingAsk() -----------------------------------------------------------
 
     @Test
     void pendingAskReturnsNullWhenNoQuestionIsOpen() throws Exception {
-        assertNull(messages.pendingAsk(T), "no async ticket at all -> no pending ask");
+        assertNull(messages.pendingAsk(T, null), "no async ticket at all -> no pending ask");
 
         String ticket = messages.sendAsync(T, "long task");
         awaitWaiting();
-        assertNull(messages.pendingAsk(T), "a plain pending delegation is not a question");
+        assertNull(messages.pendingAsk(T, null), "a plain pending delegation is not a question");
 
         injectDelivery();
         assertTrue(rendezvous.resolve(T, "done"));
         awaitTicketPhase(ticket, MessageService.Phase.DONE);
-        assertNull(messages.pendingAsk(T), "a finished ticket carries no open question either");
+        assertNull(messages.pendingAsk(T, null), "a finished ticket carries no open question either");
     }
 
     @Test
@@ -1830,7 +1901,7 @@ class MessageServiceTest {
                 CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
         MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
 
-        MessageService.PendingAsk pending = messages.pendingAsk(T);
+        MessageService.PendingAsk pending = messages.pendingAsk(T, null);
         assertNotNull(pending, "fleet_status should see the open question");
         assertEquals(ticket, pending.ticket());
         assertEquals("which config file?", pending.question());
@@ -1839,7 +1910,69 @@ class MessageServiceTest {
         CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
                 () -> messages.answer(asking.turnId(), "config.yaml", 5000));
         assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-        assertNull(messages.pendingAsk(T), "an answered question is no longer pending");
+        assertNull(messages.pendingAsk(T, null), "an answered question is no longer pending");
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
+    }
+
+    /**
+     * A caller's own terminal must match the terminal that created the delegation to see its
+     * pending question; a different terminal-bearing caller sees nothing, and a caller with no
+     * terminal at all (the unnamed primary) always sees it.
+     */
+    @Test
+    void pendingAskGatesTheQuestionByTheDelegationsCreatorTerminal() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks", null, "term_creator");
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+
+        assertNull(messages.pendingAsk(T, "term_other"),
+                "a caller whose terminal did not create the delegation must not see the question");
+
+        MessageService.PendingAsk own = messages.pendingAsk(T, "term_creator");
+        assertNotNull(own, "the creating caller must see its own open question");
+        assertEquals("which config file?", own.question());
+
+        MessageService.PendingAsk unnamed = messages.pendingAsk(T, null);
+        assertNotNull(unnamed, "a caller with no terminal (the unnamed primary) must always see the question");
+        assertEquals("which config file?", unnamed.question());
+
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
+    }
+
+    /**
+     * A task created with no recorded creator terminal (a short {@code sendAsync} overload) must
+     * not hand its open question to any caller that does have a terminal — only a caller with no
+     * terminal at all may still see it.
+     */
+    @Test
+    void pendingAskDeniesATerminalBearingCallerWhenTheTaskRecordsNoCreator() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks"); // no creatorTerminal recorded
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+
+        assertNull(messages.pendingAsk(T, "term_someone"),
+                "a terminal-bearing caller must not see a question whose task records no creator");
+        assertNotNull(messages.pendingAsk(T, null),
+                "the unnamed primary must still see it even with no recorded creator");
+
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
         awaitWaiting();
         assertTrue(rendezvous.resolve(T, "done"));
         assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());

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

@@ -2,6 +2,7 @@ package dev.ltms.fleet.rest;
 
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import dev.ltms.fleet.auth.CallerResolver;
 import dev.ltms.fleet.auth.Authz;
@@ -39,6 +40,8 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -139,6 +142,257 @@ class FleetAppAuthTest {
         assertThrows(IllegalArgumentException.class, () -> FleetApp.routeAction("GET /healthz"));
     }
 
+    // --- GET /tasks/{ticket} must not be the no-check overload ----------------------------------
+
+    /**
+     * {@code GET /tasks/{ticket}} must resolve its caller the same way {@code allow(...)} does
+     * and thread that terminal into {@link MessageService#poll(String, String)}, not the
+     * no-check overload that ignores who is asking.
+     */
+    @Test
+    void theTaskStatusRouteActuallyThreadsTheCallersTerminalIntoPoll() throws Exception {
+        String source = Files.readString(REST_SOURCE);
+
+        int start = source.indexOf("private void taskStatus(Context ctx) {");
+        assertTrue(start >= 0, "could not find taskStatus in " + REST_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("private static void herdrError(Context ctx, HerdrException e) {", start);
+        assertTrue(end > start, "could not find the method declared after taskStatus to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does call messages.poll(...) -- if this fails, the
+        // anchors above moved and the assertions below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("messages.poll("),
+                "control failed: the scraped taskStatus block contains no messages.poll( call at all "
+                        + "-- the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("caller.terminal()"),
+                "the taskStatus route must thread the resolved caller's terminal into messages.poll(...), "
+                        + "not the no-check overload -- block: " + handlerBlock);
+        assertTrue(handlerBlock.contains("ctx.attribute(CALLER)"),
+                "the taskStatus route must resolve its caller the same way allow(...) does, not via a "
+                        + "second, separate resolution path -- block: " + handlerBlock);
+    }
+
+    /**
+     * {@code GET /sessions/{id}/status} must resolve its caller the same way {@code allow(...)}
+     * does and thread that terminal into {@link MessageService#pendingAsk(String, String)}, not
+     * the no-check overload that ignores who is asking.
+     */
+    @Test
+    void theSessionStatusRouteActuallyThreadsTheCallersTerminalIntoPendingAsk() throws Exception {
+        String source = Files.readString(REST_SOURCE);
+
+        int start = source.indexOf("private void sessionStatus(Context ctx) {");
+        assertTrue(start >= 0, "could not find sessionStatus in " + REST_SOURCE
+                + " -- the scrape has stopped matching, fix the anchor before trusting this test");
+        int end = source.indexOf("private void taskStatus(Context ctx) {", start);
+        assertTrue(end > start, "could not find the method declared after sessionStatus to bound the scrape");
+        String handlerBlock = source.substring(start, end);
+
+        // CONTROL: the block we scraped really does call messages.pendingAsk(...) -- if this fails,
+        // the anchors above moved and the assertions below would otherwise pass on nothing.
+        assertTrue(handlerBlock.contains("messages.pendingAsk("),
+                "control failed: the scraped sessionStatus block contains no messages.pendingAsk( "
+                        + "call at all -- the anchors have drifted, this test is not testing what it claims to");
+
+        assertTrue(handlerBlock.contains("caller.terminal()"),
+                "the sessionStatus route must thread the resolved caller's terminal into "
+                        + "messages.pendingAsk(...), not the no-check overload -- block: " + handlerBlock);
+        assertTrue(handlerBlock.contains("ctx.attribute(CALLER)"),
+                "the sessionStatus route must resolve its caller the same way allow(...) does, not via "
+                        + "a second, separate resolution path -- block: " + handlerBlock);
+    }
+
+    /**
+     * {@code GET /tasks/{ticket}} refuses a worker whose terminal did not create the ticket,
+     * while the creating worker and the unnamed primary both still read it. The ticket is minted
+     * directly on the shared {@link MessageService}, the same way {@code MessageServiceTest}
+     * drives {@link MessageService#poll(String, String)}, so this exercises only the REST poll
+     * route's own handling of the ownership already recorded on the ticket.
+     */
+    @Test
+    void restPollRefusesADifferentWorkerButAllowsTheCreatorAndTheUnnamedPrimary() throws Exception {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        Injector injector = new Injector(agents);
+        MessageService messages = new MessageService(agents, injector, new Rendezvous());
+
+        Javalin creatorApp = startOnSharedService(messages, herdr, FakeHerdr.WORKER_PID); // -> term_a
+        Javalin otherWorkerApp = startOnSharedService(messages, herdr, 9001L);             // -> term_shell
+        Javalin primaryApp = startOnSharedService(messages, herdr, 999_999L);              // no pane -> primary
+        try {
+            String ticket = messages.sendAsync("term_a", "long task", null, "term_a");
+
+            HttpResponse<String> refused = send(otherWorkerApp.port(), "GET", "/tasks/" + ticket, null, null);
+            assertEquals(200, refused.statusCode());
+            assertTrue(refused.body().contains("forbidden"),
+                    "a different worker's terminal must be refused, not shown the ticket: " + refused.body());
+            assertFalse(refused.body().contains("\"reply\""),
+                    "a refusal must never carry reply text: " + refused.body());
+
+            HttpResponse<String> own = send(creatorApp.port(), "GET", "/tasks/" + ticket, null, null);
+            assertEquals(200, own.statusCode());
+            assertFalse(own.body().contains("forbidden"),
+                    "the creating worker must read its own ticket: " + own.body());
+
+            HttpResponse<String> primary = send(primaryApp.port(), "GET", "/tasks/" + ticket, null, null);
+            assertEquals(200, primary.statusCode());
+            assertFalse(primary.body().contains("forbidden"),
+                    "the unnamed primary must read any ticket: " + primary.body());
+        } finally {
+            creatorApp.stop();
+            otherWorkerApp.stop();
+            primaryApp.stop();
+        }
+    }
+
+    /**
+     * {@code POST /sessions/{id}/message} with {@code wait:false} must record the creating
+     * caller's own terminal on the ticket it returns, so that caller can still poll its own
+     * ticket over REST, while a different terminal is refused.
+     */
+    @Test
+    void restSendAsyncRecordsTheCreatingCallersTerminalSoItCanStillPollItsOwnTicket() throws Exception {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        Injector injector = new Injector(agents);
+        MessageService messages = new MessageService(agents, injector, new Rendezvous());
+
+        Javalin leadApp = startOnSharedService(messages, herdr, FakeHerdr.WORKER_PID, Map.of("term_a", "lead-x"));
+        Javalin otherWorkerApp = startOnSharedService(messages, herdr, 9001L); // -> term_shell
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            HttpResponse<String> created = send(leadApp.port(), "POST", "/sessions/term_a/message",
+                    "{\"content\":\"long task\",\"wait\":false}", null);
+            assertEquals(202, created.statusCode(), created.body());
+            String ticket = mapper.readTree(created.body()).path("ticket").asText(null);
+            assertNotNull(ticket, "the accepted response carried no ticket: " + created.body());
+
+            HttpResponse<String> own = send(leadApp.port(), "GET", "/tasks/" + ticket, null, null);
+            assertEquals(200, own.statusCode());
+            assertFalse(own.body().contains("forbidden"),
+                    "the session that created the ticket over REST must be able to poll it: " + own.body());
+
+            HttpResponse<String> refused = send(otherWorkerApp.port(), "GET", "/tasks/" + ticket, null, null);
+            assertEquals(200, refused.statusCode());
+            assertTrue(refused.body().contains("forbidden: this ticket was created by a different session"),
+                    "a different terminal must still be refused with the ownership detail, not some "
+                            + "other rejection: " + refused.body());
+        } finally {
+            leadApp.stop();
+            otherWorkerApp.stop();
+        }
+    }
+
+    /**
+     * {@code GET /sessions/{id}/status} shows a worker's pending {@code fleet_ask} question, its
+     * {@code turnId} and its ticket only to the caller whose terminal created that delegation, or
+     * to a caller with no terminal at all (the unnamed primary) — a different terminal-bearing
+     * caller still sees the base status line, but none of the pending-ask fields.
+     */
+    @Test
+    void restStatusGatesThePendingAskFieldsByTheDelegationsCreatorTerminal() throws Exception {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        Injector injector = new Injector(agents);
+        Rendezvous rendezvous = new Rendezvous();
+        MessageService messages = new MessageService(agents, injector, rendezvous);
+
+        Javalin creatorApp = startOnSharedService(messages, herdr, FakeHerdr.WORKER_PID); // -> term_a
+        Javalin otherWorkerApp = startOnSharedService(messages, herdr, 9001L);             // -> term_shell
+        Javalin primaryApp = startOnSharedService(messages, herdr, 999_999L);              // no pane -> primary
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            String ticket = messages.sendAsync("term_target", "task that asks", null, "term_a");
+            long deadline = System.currentTimeMillis() + 3000;
+            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
+                Thread.sleep(5);
+            }
+            assertTrue(rendezvous.isWaiting("term_target"), "sendAsync should have opened its rendezvous waiter");
+
+            CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
+                    () -> messages.ask("term_target", "which config file?", 5000));
+
+            MessageService.TaskView asking;
+            deadline = System.currentTimeMillis() + 3000;
+            do {
+                asking = messages.poll(ticket, null);
+                Thread.sleep(5);
+            } while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline);
+            assertEquals(MessageService.Phase.ASKING, asking.phase());
+            String turnId = asking.turnId();
+
+            JsonNode other = mapper.readTree(
+                    send(otherWorkerApp.port(), "GET", "/sessions/term_target/status", null, null).body());
+            assertEquals("idle", other.get("status").asText(), "the base status must still be shown");
+            assertFalse(other.has("question"), "a non-creating caller must not see the question: " + other);
+            assertFalse(other.has("turnId"), "a non-creating caller must not see the turnId: " + other);
+            assertFalse(other.has("ticket"), "a non-creating caller must not see the ticket: " + other);
+
+            JsonNode own = mapper.readTree(
+                    send(creatorApp.port(), "GET", "/sessions/term_target/status", null, null).body());
+            assertEquals("which config file?", own.get("question").asText(), "the creator must see the question");
+            assertEquals(turnId, own.get("turnId").asText(), "the creator must see the turnId");
+            assertEquals(ticket, own.get("ticket").asText(), "the creator must see the ticket");
+
+            JsonNode primary = mapper.readTree(
+                    send(primaryApp.port(), "GET", "/sessions/term_target/status", null, null).body());
+            assertEquals("which config file?", primary.get("question").asText(),
+                    "a caller with no terminal (the unnamed primary) must see the question");
+
+            // Clean up the still-open ask so the background thread does not linger past the test.
+            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                    () -> messages.answer(turnId, "config.yaml", 5000));
+            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+            deadline = System.currentTimeMillis() + 3000;
+            while (!rendezvous.isWaiting("term_target") && System.currentTimeMillis() < deadline) {
+                Thread.sleep(5);
+            }
+            assertTrue(rendezvous.resolve("term_target", "done"));
+            answer.get(5, TimeUnit.SECONDS);
+        } finally {
+            creatorApp.stop();
+            otherWorkerApp.stop();
+            primaryApp.stop();
+        }
+    }
+
+    /**
+     * As {@link #start}, but shares {@code messages} and {@code herdr} across several app
+     * instances bound to different pids, each returned as its own started {@link Javalin} rather
+     * than through the shared {@code app} field, so several differently-resolved callers can
+     * poll the same ticket.
+     */
+    private Javalin startOnSharedService(MessageService messages, FakeHerdr herdr, long pid) {
+        return startOnSharedService(messages, herdr, pid, Map.of());
+    }
+
+    /**
+     * As {@link #startOnSharedService(MessageService, FakeHerdr, long)}, but {@code leadTerminals}
+     * resolves the given pid's terminal to a named lead (a caller with SEND permission) instead of
+     * a plain worker, for a test that needs a terminal-bearing caller able to create a ticket.
+     */
+    private Javalin startOnSharedService(MessageService messages, FakeHerdr herdr, long pid,
+            Map<String, String> leadTerminals) {
+        FleetConfig.Profile wcfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        AgentControl agents = new AgentControl(herdr);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(
+                agents, new WorkspaceControl(herdr), new SubscriptionGuard(Set.of("gx00.gw")),
+                Map.of(wcfg.profile(), wcfg), wcfg.profile(),
+                k -> "FLEETD_WORKER_TOKEN".equals(k) ? "tok-abc" : null);
+        SessionManager sessions = new SessionManager(workers, new FakeWorktrees());
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> pid);
+        CallerResolver callers = CallerResolver.withLeadsAndMembers(identity, false, null,
+                () -> leadTerminals, new MemberRegistry(null));
+        Metrics appMetrics = FleetMetrics.create(sessions, new dev.ltms.fleet.msg.InMemoryReplyInbox());
+
+        return new FleetApp(herdr, workers, sessions, messages, sessions.asPresence(), null,
+                callers, appMetrics).build().start("127.0.0.1", 0);
+    }
+
     /**
      * fleetd #669 Unit A: {@code POST /sessions/{id}/message} is two call shapes behind one route,
      * mirroring {@code fleet_send}'s MCP-side split into {@link Authz.Action#SEND} and {@link
@@ -205,16 +459,72 @@ class FleetAppAuthTest {
 
     /**
      * {@code permitsFor} is the exact decision {@link FleetApp#allow} makes, passing the real
-     * production classifier rather than a test-supplied one — no terminal is recognised as a
-     * configured lead or collaborator, so a collaborator's SEND is refused through the REST gate.
+     * production classifier rather than a test-supplied one — built from an empty {@link
+     * CallerResolver}, so no terminal is recognised as a configured lead or collaborator and a
+     * collaborator's SEND is refused through the REST gate.
      */
     @Test
     void aCollaboratorMayNotSendOverRestWithTheRealProductionClassifier() {
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(new FakeHerdr()), _ -> 700L);
+        CallerResolver callers = CallerResolver.withLeadsAndMembers(identity, false, null,
+                Map::of, new MemberRegistry(null));
         Principal collaborator = Principal.collaborator("ops", "term_collab", 700);
-        assertFalse(FleetApp.permitsFor(collaborator, Authz.Action.SEND, "term_lead"),
+        assertFalse(FleetApp.permitsFor(collaborator, Authz.Action.SEND, "term_lead",
+                        callers.knownLeadOrCollaborator()),
                 "no terminal is recognised as a lead or collaborator yet");
     }
 
+    /**
+     * fleetd #669 Unit D: wires a real {@link CallerResolver} with a known lead and a known
+     * collaborator tab, and a spawned member's own terminal recognised by neither map. A
+     * collaborator's SEND reaches the known lead and the known collaborator, and is refused for the
+     * spawned member's terminal — over the REST route, not just the unit-level classifier, so a
+     * test covering only MCP cannot leave this route open.
+     */
+    @Test
+    void aCollaboratorMaySendToAKnownLeadOrCollaboratorButNotToASpawnedMembersTerminalOverRest() throws Exception {
+        int port = startWithRealClassifier(FakeHerdr.WORKER_PID,
+                Map.of("term_lead_known", "lead-x"), Map.of("term_a", "ops2"));
+
+        HttpResponse<String> toLead = send(port, "POST", "/sessions/term_lead_known/message",
+                "{\"content\":\"hi\",\"wait\":false}", null);
+        assertEquals(202, toLead.statusCode(), toLead.body());
+
+        HttpResponse<String> toSpawnedMembersTerminal = send(port, "POST", "/sessions/term_worker/message",
+                "{\"content\":\"hi\",\"wait\":false}", null);
+        assertEquals(403, toSpawnedMembersTerminal.statusCode(), toSpawnedMembersTerminal.body());
+    }
+
+    /**
+     * As {@link #start}, but with explicit lead/collaborator maps and no spawned-member roster, so
+     * a test can wire the real {@link CallerResolver#knownLeadOrCollaborator()} classifier instead
+     * of the default empty one.
+     */
+    private int startWithRealClassifier(long pid, Map<String, String> leadTerminals,
+                                        Map<String, String> collaboratorTerminals) {
+        FakeHerdr herdr = new FakeHerdr();
+        FleetConfig.Profile wcfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN", null,
+                "tab", "fleetd-workers", "worker: {profile} #{n}", null, null, null);
+        AgentControl agents = new AgentControl(herdr);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(
+                agents, new WorkspaceControl(herdr), new SubscriptionGuard(Set.of("gx00.gw")),
+                Map.of(wcfg.profile(), wcfg), wcfg.profile(),
+                k -> "FLEETD_WORKER_TOKEN".equals(k) ? "tok-abc" : null);
+        SessionManager sessions = new SessionManager(workers, new FakeWorktrees());
+        Injector injector = new Injector(agents);
+        MessageService messages = new MessageService(agents, injector, new Rendezvous());
+
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> pid);
+        CallerResolver callers = CallerResolver.withLeadsAndMembers(identity, false, null,
+                () -> leadTerminals, new MemberRegistry(null), t -> null, () -> collaboratorTerminals);
+        metrics = FleetMetrics.create(sessions, new dev.ltms.fleet.msg.InMemoryReplyInbox());
+
+        app = new FleetApp(herdr, workers, sessions, messages, sessions.asPresence(), null,
+                callers, metrics).build().start("127.0.0.1", 0);
+        return app.port();
+    }
+
     // --- loopback-trust: the caller is the primary -------------------------------------------
 
     @Test

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

@@ -2436,4 +2436,200 @@ class SessionManagerTest {
             return "threw:" + e.getClass().getName() + ":" + e.getMessage();
         }
     }
+
+    // --- fleetd #702: spawnedMemberRole must still answer for a pane mid-teardown ----------------
+
+    /**
+     * A {@link Worktrees} test double whose {@code hasUncommitted} runs an injected hook before
+     * answering. This is what lets a test resolve a releasing pane's terminal from inside the
+     * window {@link SessionManager#release} opens between removing the registry entry and
+     * actually stopping the pane — the hook runs synchronously on the release call's own thread,
+     * at the exact point {@code release} shells out to {@code git status}, so there is no sleep
+     * and no race to land in it.
+     */
+    private static final class HookedWorktrees implements Worktrees {
+        private Runnable hook;
+        private boolean dirty = false;
+
+        HookedWorktrees onHasUncommitted(Runnable hook) {
+            this.hook = hook;
+            return this;
+        }
+
+        @Override
+        public String add(String repoRoot, String branch, String baseRef) {
+            return "/wt/" + branch.replace('/', '_');
+        }
+
+        @Override
+        public void remove(String repoRoot, String worktreePath) {
+        }
+
+        @Override
+        public void deleteBranch(String repoRoot, String branch) {
+        }
+
+        @Override
+        public boolean hasUncommitted(String worktreePath) {
+            if (hook != null) {
+                hook.run();
+            }
+            return dirty;
+        }
+
+        @Override
+        public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
+        }
+
+        @Override
+        public String repoRoot(String cwd) {
+            return "/repo";
+        }
+
+        @Override
+        public java.util.Optional<String> snapshot(String worktreePath, String branch, String message) {
+            return java.util.Optional.empty();
+        }
+
+        @Override
+        public WipRefStats wipRefs(String repoRoot) {
+            return new WipRefStats(0, 0L);
+        }
+
+        @Override
+        public int pruneWipRefs(String repoRoot, long minAgeMillis) {
+            return 0;
+        }
+
+        @Override
+        public void shareWithGroup(String repoRoot, String worktreePath) {
+        }
+    }
+
+    @Test
+    void spawnedMemberRoleResolvesTheLiveRegisteredRole() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller", null);
+
+        assertEquals(s.role(), sessions.spawnedMemberRole(s.terminalId()),
+                "a registered session resolves to its own role");
+        assertNull(sessions.spawnedMemberRole("term_unknown"),
+                "a terminal with no session at all resolves to null");
+    }
+
+    @Test
+    void spawnedMemberRoleIsNullOnceReleaseFullyCompletes() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr, new HookedWorktrees());
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702a", null));
+
+        sessions.release(s.paneId());
+
+        assertNull(sessions.spawnedMemberRole(s.terminalId()),
+                "once release has fully finished, the terminal is neither registered nor releasing");
+    }
+
+    @Test
+    void spawnedMemberRoleStillAnswersBetweenTheRegistryRemovalAndThePaneStop() {
+        FakeHerdr herdr = new FakeHerdr();
+        java.util.concurrent.atomic.AtomicReference<MemberRole> duringWindow = new java.util.concurrent.atomic.AtomicReference<>();
+        HookedWorktrees worktrees = new HookedWorktrees();
+        SessionManager sessions = sessionManager(herdr, worktrees);
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("fleetd-702b", null));
+        worktrees.onHasUncommitted(() -> {
+            // This runs from INSIDE release()'s git-status shell-out: the registry entry is
+            // already gone, but the pane has not stopped yet — a real call landing inside the
+            // exact window fleetd #702 reports, so no sleep and no race is needed to reach it.
+            assertTrue(sessions.get(s.paneId()).isEmpty(),
+                    "sanity: the registry entry is already gone at this point");
+            duringWindow.set(sessions.spawnedMemberRole(s.terminalId()));
+        });
+
+        sessions.release(s.paneId());
+
+        assertEquals(s.role(), duringWindow.get(),
+                "spawnedMemberRole must still answer the live role while the pane is mid-teardown, "
+                        + "not only while the session is still in the registry");
+        assertNull(sessions.spawnedMemberRole(s.terminalId()),
+                "and once release has fully finished, the window is closed too");
+    }
+
+    /**
+     * {@code Releasing} is private, so its {@code enter}/{@code leave} are reached through
+     * reflection. Each is called directly, never through {@link SessionManager#release} or
+     * {@link SessionManager#spawnedMemberRole}, so a test naming one of them exercises only that
+     * one — a regression in the other can never hide behind it.
+     */
+    private static Object releasingOf(int depth, String terminalId, MemberRole role) throws ReflectiveOperationException {
+        Class<?> cls = Class.forName("dev.ltms.fleet.session.SessionManager$Releasing");
+        java.lang.reflect.Constructor<?> ctor = cls.getDeclaredConstructor(int.class, String.class, MemberRole.class);
+        ctor.setAccessible(true);
+        return ctor.newInstance(depth, terminalId, role);
+    }
+
+    private static Object releasingEnter(Object prior, MemberSession known) throws ReflectiveOperationException {
+        Class<?> cls = Class.forName("dev.ltms.fleet.session.SessionManager$Releasing");
+        java.lang.reflect.Method m = cls.getDeclaredMethod("enter", cls, MemberSession.class);
+        m.setAccessible(true);
+        return m.invoke(null, prior, known);
+    }
+
+    private static Object releasingLeave(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("leave");
+        m.setAccessible(true);
+        return m.invoke(releasing);
+    }
+
+    private static int depthOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("depth");
+        m.setAccessible(true);
+        return (int) m.invoke(releasing);
+    }
+
+    private static String terminalIdOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("terminalId");
+        m.setAccessible(true);
+        return (String) m.invoke(releasing);
+    }
+
+    private static MemberRole roleOf(Object releasing) throws ReflectiveOperationException {
+        java.lang.reflect.Method m = releasing.getClass().getDeclaredMethod("role");
+        m.setAccessible(true);
+        return (MemberRole) m.invoke(releasing);
+    }
+
+    @Test
+    void releasingLeaveStepsDownADepthGreaterThanOneInsteadOfRemovingIt() throws ReflectiveOperationException {
+        Object depthTwo = releasingOf(2, "term_a", MemberRole.DEV);
+
+        Object afterLeave = releasingLeave(depthTwo);
+
+        assertNotNull(afterLeave,
+                "depth 2 means another release of the SAME pane is still mid-teardown; leave() must "
+                        + "step the depth down, never remove the marker outright — removing it here "
+                        + "is what a plain Set would do, and would reopen the window the still-in-"
+                        + "flight release is relying on staying closed");
+        assertEquals(1, depthOf(afterLeave));
+        assertEquals("term_a", terminalIdOf(afterLeave));
+        assertEquals(MemberRole.DEV, roleOf(afterLeave));
+    }
+
+    @Test
+    void releasingEnterPreservesThePriorTerminalWhenTheOverlappingCallHasNoSessionOfItsOwn()
+            throws ReflectiveOperationException {
+        Object prior = releasingOf(1, "term_a", MemberRole.DEV);
+
+        Object afterEnter = releasingEnter(prior, null);
+
+        assertEquals(2, depthOf(afterEnter), "depth still increments whether or not this enter knows its session");
+        assertEquals("term_a", terminalIdOf(afterEnter),
+                "a losing CAS (the releaseIfCurrent race fleetd #702 is about) has no session of "
+                        + "its own to pass as known, and must not blank out the terminal the winning "
+                        + "call already recorded — that terminal is what spawnedMemberRole matches "
+                        + "against");
+        assertEquals(MemberRole.DEV, roleOf(afterEnter));
+    }
 }