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

HerdrRouter.agentsFor picked the member daemon for any terminal the lead
predicate did not recognize, so a configured collaborator's pane (opened
by a person, exactly like a lead's) was routed to the member herdr
daemon instead of the lead one.

FleetdAssembly now combines the leads map and the collaborator-terminals
map into the predicate it hands HerdrRouter. HerdrRouter's isLead field
and constructor parameter are renamed to routeToLead, with its javadoc
naming the real contract: true for any terminal whose pane lives in the
lead daemon, lead or collaborator.

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