Merge PR #704: fleetd #669 Unit E — route a collaborator's pane to the lead herdr daemon

A configured collaborator's terminal now routes to the lead herdr daemon
rather than the member one. FleetdAssembly ORs a second terminal map into
the predicate it hands HerdrRouter, and the predicate's field is renamed
isLead -> routeToLead so its name matches the widened contract.

Verified in the lead's own throwaway worktree: mvn clean install green at
1992 tests, 0 failures (174 surefire report files, fresh worktree). The
fix rests on one test, so I reproduced the mutation myself rather than
taking the worker's word: dropping the collaborator clause from the
predicate turns FleetdAssemblyCollaboratorHerdrRoutingTest red at runtime
with two distinct AgentControl identities, while HerdrRouterTest's 4 tests
stay green. That confirms both the kill and that the two-distinct-fakes
setup is actually discriminating.

Also checked by hand: both terminal maps come from LeadTabScanner's one
byKind helper and are keyed terminal_id -> name, so containsKey(target) is
correct for both; no agentsFor call runs between router construction and
the ref being set, so the publication window is harmless and matches the
one leadsRef already had.

One reviewer fanned out against the diff, briefed from the diff rather
than the implementer's rationale. It reported no issue.

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

@@ -143,8 +143,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.
@@ -292,6 +296,7 @@ final class FleetdAssembly {
             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

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/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/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");
+    }
 }