fleetd: thread the creating caller's terminal into the REST fire-and-poll send path

sendMessage's wait:false branch now records the resolved caller's own
terminal as the ticket's creatorTerminal, the same way taskStatus already
resolves its caller, so a REST-created ticket's own creator can still poll
it under the ownership check that now gates GET /tasks/{ticket}.

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:

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

@@ -249,17 +249,6 @@ public final class CallerResolver {
         return leadTerminals.get();
     }
 
-    /**
-     * The currently-recognised architect slots, {@code terminal_id → slot name} (CB-548).
-     *
-     * <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()}.
-     */
-    public Map<String, String> members() {
-        return architectTerminals.get();
-    }
-
     /**
      * The currently-recognised collaborator tabs, {@code terminal_id → name}.
      *

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/mcp/FleetMcp.java

@@ -490,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
@@ -524,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.
@@ -560,6 +560,7 @@ 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)));
                 };
@@ -743,6 +744,21 @@ 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();
+    }
+
     /** 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);
@@ -956,6 +972,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");
         }
@@ -963,7 +990,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);
     }
 
@@ -1142,9 +1169,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);
         }
@@ -1158,7 +1200,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)");
         }
@@ -1761,7 +1803,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);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1770,7 +1813,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);
     }
 
     /**
@@ -1787,7 +1831,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);
     }
 
     /** As above, plus fleetd #201 Unit 5 cool-off facts (see {@link OutageSource}). */
@@ -1796,7 +1841,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);
     }
 
     /**
@@ -1818,7 +1864,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);
     }
 
     /**
@@ -1842,7 +1889,8 @@ public final class FleetMcp {
                                                 LeadSeatSource leadSeats, Map<String, String> leads, String selfTerm,
                                                 CoordinationSource coordination, boolean callerIsPrimary) {
         return listFleet(workers, sessions, messages, capacity, healthCoverage, LoopHealthSource.none(), quarantine,
-                outage, leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm, coordination, callerIsPrimary);
+                outage, leadSeats, new LeadContextGauge(), LeadConfigDirSource.none(), leads, selfTerm,
+                Map.of(), false, coordination, callerIsPrimary);
     }
 
     /**
@@ -1852,6 +1900,13 @@ 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}
      */
     static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
                                                  CapacitySource capacity, HealthCoverageSource healthCoverage,
@@ -1860,6 +1915,7 @@ public final class FleetMcp {
                                                LeadSeatSource leadSeats, LeadContextGauge contextGauge,
                                                LeadConfigDirSource leadConfigDirs,
                                                Map<String, String> leads, String selfTerm,
+                                               Map<String, String> collaborators, boolean collaboratorsVisible,
                                                CoordinationSource coordination, boolean callerIsPrimary) {
         try {
             Map<String, Agent> live = workers.list().stream()
@@ -1886,6 +1942,16 @@ public final class FleetMcp {
             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.
@@ -2197,6 +2263,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
@@ -2422,7 +2502,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.

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

@@ -694,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;
         }
@@ -894,7 +895,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/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

@@ -460,7 +460,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

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/LeadTabScannerTest.java

@@ -205,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() {
@@ -219,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")

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

@@ -366,6 +366,85 @@ 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);
+    }
+
+    /**
+     * {@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);
+    }
+
     // --- 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);
     }
 }

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;
@@ -879,6 +880,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);
+    }
+
+    /**
+     * 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

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");

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

@@ -139,6 +139,154 @@ 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 /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();
+        }
+    }
+
+    /**
+     * 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