CB-530..536, CB-538 groundwork: leads as peers, and a fleet that can find itself

Two leads now work as peers rather than one primary plus workers. The arc: CB-530/531 lead identity: `leaders:` names panes, `leadScan:` discovers them by tab label (LeadTabScanner, TTL-cached, worker spaces excluded). CB-532 leads can message each other AND be answered. Principal.leader now carries its terminal, so ownsSession() can be true for a lead; the "and you must be a worker" conjunct beside it protected nothing. Retires `primary:` — reply nudges follow the delegating lead, a binding recorded at bridge_send where both halves are known. CB-533 ClaudeCodeLauncher passes --model. argv is usually a wrapper (`ccs <profile>`) that re-exports its own model family, so ANTHROPIC_MODEL alone was silently overruled. CB-534 a lead is deliverable. The CB-113 readiness gate only opened for terminals in WorkerPresence, which only workers ever enter, so every lead->lead send waited out the ~60s grace and failed having never been typed. The gate guards a *spawned* peer's boot window; a lead is never spawned. CB-535 bridge_list returns `leads` alongside `workers`, with `self` on the caller's row. An empty worker roster no longer reads as "no peers". CB-536 CLAUDE.md: lead<->lead is coordinate-only, never sideways delegation. Propagated byte-identically to wiki/7-Use-Cases.md. MIXED PROVENANCE — recorded deliberately rather than hidden. This tree also carries in-progress CB-537 (context separation) authored by the peer lead gpt-sol-5.6 and its worker: Capability.CONTEXT_RESET, SessionManager.clearAfterTurn, and the Injector/TurnListener/CompletionResolver/launcher changes around it. That work was done in this shared working tree rather than a worktree, and is entangled with the above in BridgedConfig.java, Bridged.java and ClaudeCodeLauncher.java, so neither lead could stage its own half without sweeping in the other's. Committing the whole green state is the honest resolution; the peer branches from here. Note for whoever picks CB-537 up: the design in this commit is SUPERSEDED. Both leads agreed to replace the global `clearAfterTurn` boolean with per-delivery policy (inherit|fresh|thread) applied PRE-delivery, because a post-turn reset races by construction — Injector.onStatus clears awaitingCompletion and dequeues the next message in the same tick. `fresh` is also a correctness guarantee, so an adapter without a reset capability must refuse it rather than log a no-op. mvn clean install: Tests run: 464, Failures: 0, Errors: 0, Skipped: 0. BUILD SUCCESS.
2026-08-13 09:38:24 +02:00
parent ef1e014b41
commit bc13b8e92c
34 changed files with 2350 additions and 115 deletions
@@ -0,0 +1,86 @@
+package dev.ltms.bridged;
+
+import dev.ltms.bridged.inject.WorkerPresence;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+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.
+ *
+ * <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.
+ */
+class BridgedDeliverabilityTest {
+
+    private static Supplier<Map<String, String>> leads(Map<String, String> m) {
+        return () -> m;
+    }
+
+    @Test
+    @DisplayName("a worker that has connected its MCP is deliverable")
+    void presentWorkerIsDeliverable() {
+        WorkerPresence presence = new WorkerPresence();
+        presence.markPresent("term_worker");
+
+        assertTrue(Bridged.deliverableTo(presence, leads(Map.of())).test("term_worker"));
+    }
+
+    @Test
+    @DisplayName("a worker still in its boot window is held back")
+    void absentWorkerIsNotDeliverable() {
+        assertFalse(Bridged.deliverableTo(new WorkerPresence(), leads(Map.of())).test("term_booting"));
+    }
+
+    @Test
+    @DisplayName("a lead is deliverable without ever being marked present")
+    void leadIsDeliverableWithoutPresence() {
+        WorkerPresence presence = new WorkerPresence();
+        Predicate<String> deliverable =
+                Bridged.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")));
+
+        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")
+    void strangerIsNotDeliverable() {
+        WorkerPresence presence = new WorkerPresence();
+        presence.markPresent("term_worker");
+
+        assertFalse(Bridged.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")))
+                .test("term_stranger"));
+    }
+
+    @Test
+    @DisplayName("a lead discovered after startup becomes deliverable with no restart")
+    void leadSetIsReadThroughOnEveryCall() {
+        Map<String, String> discovered = new HashMap<>();
+        Predicate<String> deliverable = Bridged.deliverableTo(new WorkerPresence(), leads(discovered));
+
+        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("forgetting a torn-down worker does not strip a lead of its deliverability")
+    void forgetDoesNotDisarmALead() {
+        WorkerPresence presence = new WorkerPresence();
+        Predicate<String> deliverable =
+                Bridged.deliverableTo(presence, leads(Map.of("term_lead", "opus-5.0")));
+
+        presence.forget("term_lead"); // the injector's cleanup path runs against every target
+
+        assertTrue(deliverable.test("term_lead"));
+    }
+}
@@ -5,6 +5,8 @@ import dev.ltms.bridged.herdr.PaneLocator;
 import dev.ltms.bridged.mcp.ConnectionIdentity;
 import org.junit.jupiter.api.Test;

+import java.util.Map;
+
 import static org.junit.jupiter.api.Assertions.*;

 /**
@@ -48,7 +50,7 @@ class CallerResolverTest {
    void aPinnedPrimaryTerminalResolvesToPrimaryNotWorker() {
        // The primary's own session lives in a herdr pane (term_a here). Without the pin the pane
        // match wins and the primary is locked out of spawn/send/stop as a misread worker.
-        Principal p = new CallerResolver(workerIdentity(), false, null, "term_a")
+        Principal p = CallerResolver.pinnedTo(workerIdentity(), false, null, "term_a")
                .resolve("127.0.0.1", 42, null);

        assertEquals(Role.PRIMARY, p.role());
@@ -56,7 +58,7 @@ class CallerResolverTest {

    @Test
    void aPinnedPrimaryTerminalNeedsNoTokenEvenInTokenMode() {
-        Principal p = new CallerResolver(workerIdentity(), true, "s3cret", "term_a")
+        Principal p = CallerResolver.pinnedTo(workerIdentity(), true, "s3cret", "term_a")
                .resolve("127.0.0.1", 42, null);

        assertEquals(Role.PRIMARY, p.role(),
@@ -65,7 +67,7 @@ class CallerResolverTest {

    @Test
    void otherPanesRemainWorkersWhenAPinIsSet() {
-        Principal p = new CallerResolver(workerIdentity(), false, null, "term_someone_else")
+        Principal p = CallerResolver.pinnedTo(workerIdentity(), false, null, "term_someone_else")
                .resolve("127.0.0.1", 42, null);

        assertEquals(Role.WORKER, p.role());
@@ -76,9 +78,9 @@ class CallerResolverTest {
    @Test
    void aBlankPinLeavesWorkerResolutionUntouched() {
        assertEquals(Role.WORKER,
-                new CallerResolver(workerIdentity(), false, null, "  ").resolve("127.0.0.1", 42, null).role());
+                CallerResolver.pinnedTo(workerIdentity(), false, null, "  ").resolve("127.0.0.1", 42, null).role());
        assertEquals(Role.WORKER,
-                new CallerResolver(workerIdentity(), false, null, null).resolve("127.0.0.1", 42, null).role());
+                CallerResolver.pinnedTo(workerIdentity(), false, null, null).resolve("127.0.0.1", 42, null).role());
    }

    @Test
@@ -133,6 +135,169 @@ class CallerResolverTest {
        assertEquals(Role.ANONYMOUS, p.role());
    }

+    // ── CB-530: the leaders registry ────────────────────────────────────────────────────────────
+    // The fake resolves exactly one pane (term_a) from a PID, so "two leads both resolve" is
+    // asserted at the config layer (BridgedConfigTest#leaderTerminals…). What matters here is that
+    // resolution is a REGISTRY LOOKUP rather than a single equality test against one pin.
+
+    @Test
+    void aRegisteredLeadPaneResolvesToPrimaryCarryingItsName() {
+        Principal p = new CallerResolver(workerIdentity(), false, null, Map.of("term_a", "opus-5.0"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.PRIMARY, p.role());
+        assertEquals("opus-5.0", p.name(), "whoami must be able to say WHICH lead is asking");
+        assertEquals("term_a", p.terminal(),
+                "CB-532: a lead carries the pane it was matched by. Without it ownsSession() can "
+                        + "never be true for a lead, so it can send to a peer but never answer one");
+    }
+
+    @Test
+    void aSecondLeadIsRecognisedRatherThanSilentlyDemoted() {
+        // The regression this feature exists for: with a singular pin, whichever lead was not the
+        // pin resolved as a worker and was refused every orchestration call.
+        Principal p = new CallerResolver(workerIdentity(), false, null,
+                Map.of("term_elsewhere", "gpt-sol-5.6", "term_a", "opus-5.0"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.PRIMARY, p.role());
+        assertEquals("opus-5.0", p.name());
+    }
+
+    @Test
+    void aPaneAbsentFromTheRegistryIsStillAWorker() {
+        Principal p = new CallerResolver(workerIdentity(), false, null,
+                Map.of("term_elsewhere", "gpt-sol-5.6"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.WORKER, p.role());
+        assertEquals("term_a", p.terminal());
+        assertNull(p.name());
+    }
+
+    @Test
+    void aRegisteredLeadNeedsNoTokenEvenInTokenMode() {
+        Principal p = new CallerResolver(workerIdentity(), true, "s3cret", Map.of("term_a", "opus"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.PRIMARY, p.role(),
+                "the pane mapping is as unforgeable as a worker's — it outranks the token path");
+        assertEquals("opus", p.name());
+    }
+
+    /** The pre-CB-530 spelling must keep working, exactly, including for configs that never migrate. */
+    @Test
+    void theLegacySinglePinBehavesAsALeadNamedPrimary() {
+        Principal p = CallerResolver.pinnedTo(workerIdentity(), false, null, "term_a")
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.PRIMARY, p.role());
+        assertEquals("primary", p.name());
+    }
+
+    @Test
+    void anEmptyRegistryLeavesEveryPaneAWorker() {
+        Map<String, String> noLeads = null;
+        assertEquals(Role.WORKER,
+                new CallerResolver(workerIdentity(), false, null, Map.of())
+                        .resolve("127.0.0.1", 42, null).role());
+        assertEquals(Role.WORKER,
+                new CallerResolver(workerIdentity(), false, null, noLeads)
+                        .resolve("127.0.0.1", 42, null).role());
+        // CB-531: and the same for the live-registry form, whose supplier may also be absent.
+        assertEquals(Role.WORKER,
+                CallerResolver.withLeads(workerIdentity(), false, null, null)
+                        .resolve("127.0.0.1", 42, null).role());
+    }
+
+    /** The audit line must distinguish leads once several exist, or a log says nothing useful. */
+    @Test
+    void describeNamesTheLeadButStillReadsPrimaryWhenUnnamed() {
+        assertEquals("leader:opus-5.0", Principal.leader("opus-5.0", "term_a", 1).describe());
+        assertEquals("primary", Principal.primary(1).describe());
+        assertEquals("worker:term_a", Principal.worker("term_a", 1).describe());
+    }
+
+    // ── CB-532: a lead is an addressable peer, not only a sender ────────────────────────────────
+
+    /**
+     * The regression this ticket exists for: two leads could both be recognised (CB-530/531) and
+     * still not converse, because REPLY is gated on ownsSession() and a lead owned nothing.
+     */
+    @Test
+    void aLeadOwnsItsOwnPaneSoItMayAnswerAPeer() {
+        Principal lead = new CallerResolver(workerIdentity(), false, null, Map.of("term_a", "opus-5.0"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertTrue(lead.ownsSession("term_a"));
+        assertTrue(Authz.permits(lead, Authz.Action.REPLY, "term_a"),
+                "a lead answering a peer replies for its OWN terminal — the rendezvous the sender "
+                        + "opened is keyed on exactly that");
+        assertTrue(Authz.permits(lead, Authz.Action.ASK, "term_a"));
+    }
+
+    @Test
+    void aLeadStillCannotActAsAnyoneElse() {
+        Principal lead = new CallerResolver(workerIdentity(), false, null, Map.of("term_a", "opus-5.0"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertFalse(lead.ownsSession("term_someone_else"));
+        assertFalse(Authz.permits(lead, Authz.Action.REPLY, "term_someone_else"),
+                "widening WHO may reply must not widen WHAT they may reply as");
+    }
+
+    /** A primary with no pane — token mode, or off-host — owns nothing and must stay a sender only. */
+    @Test
+    void anUnnamedPrimaryWithNoPaneOwnsNothing() {
+        Principal p = new CallerResolver(nonWorkerIdentity(), true, "s3cret")
+                .resolve("127.0.0.1", 99, "Bearer s3cret");
+
+        assertEquals(Role.PRIMARY, p.role());
+        assertNull(p.terminal());
+        assertFalse(p.ownsSession(null), "a null terminal must never match a null session id");
+        assertFalse(Authz.permits(p, Authz.Action.REPLY, null));
+    }
+
+    @Test
+    void aLeadKeepsEveryOrchestrationRightItAlreadyHad() {
+        Principal lead = new CallerResolver(workerIdentity(), false, null, Map.of("term_a", "opus-5.0"))
+                .resolve("127.0.0.1", 42, null);
+
+        assertTrue(Authz.permits(lead, Authz.Action.SPAWN, null));
+        assertTrue(Authz.permits(lead, Authz.Action.SEND, "term_worker"));
+        assertTrue(Authz.permits(lead, Authz.Action.STOP, null));
+        assertTrue(Authz.permits(lead, Authz.Action.DRAIN, null));
+    }
+
+    /**
+     * CB-531: the registry is read per resolve, not snapshotted at construction — a lead that
+     * labels its tab after the daemon booted is recognised without a restart.
+     */
+    @Test
+    void aLeadRegisteredAfterConstructionIsHonouredWithoutRebuildingTheResolver() {
+        Map<String, String> live = new java.util.HashMap<>();
+        CallerResolver r = CallerResolver.withLeads(workerIdentity(), false, null, () -> live);
+
+        assertEquals(Role.WORKER, r.resolve("127.0.0.1", 42, null).role());
+
+        live.put("term_a", "gpt-sol-5.6"); // the scanner sees a newly-labelled tab
+
+        Principal p = r.resolve("127.0.0.1", 42, null);
+        assertEquals(Role.PRIMARY, p.role());
+        assertEquals("gpt-sol-5.6", p.name());
+    }
+
+    /** The map form must stay a snapshot: a caller handing over a map is not offering live state. */
+    @Test
+    void theMapFormIsCopiedSoLaterMutationCannotGrantLeadership() {
+        Map<String, String> mutable = new java.util.HashMap<>();
+        CallerResolver r = new CallerResolver(workerIdentity(), false, null, mutable);
+
+        mutable.put("term_a", "sneaky");
+
+        assertEquals(Role.WORKER, r.resolve("127.0.0.1", 42, null).role());
+    }
+
    @Test
    void tokenModeRequiresANonEmptyConfiguredToken() {
        ConnectionIdentity id = nonWorkerIdentity();
@@ -5,6 +5,8 @@ import org.junit.jupiter.api.io.TempDir;

 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
 import java.util.Set;

 import static org.junit.jupiter.api.Assertions.*;
@@ -45,6 +47,7 @@ class BridgedConfigTest {
        assertEquals(9000, cfg.bind().port());
        assertNotNull(cfg.guard(), "guard must default to empty, never null");
        assertTrue(cfg.guard().offSubscriptionHosts().isEmpty());
+        assertFalse(cfg.lifecycle().clearAfterTurn(), "context clearing is opt-in");
    }

    @Test
@@ -96,6 +99,233 @@ class BridgedConfigTest {
        assertDoesNotThrow(() -> BridgedConfig.load(f));
    }

+    /**
+     * CB-530. Unknown keys stay ignored — config must be allowed to run ahead of the code — but they
+     * must be NAMED at load. A whole block that parses, is dropped, and is never mentioned again is
+     * indistinguishable from one that works: that is exactly how a hand-written `leaders:` registry
+     * came to look configured while being inert.
+     */
+    @Test
+    void unknownTopLevelKeysAreNamedSoADroppedBlockCannotLookLikeAWorkingOne() {
+        assertEquals(List.of("futureFeature", "leedars"),
+                BridgedConfig.unknownTopLevelKeys(
+                        "bind:\n  port: 8080\nleedars:\n  a: b\nfutureFeature: true\n"),
+                "a typo'd key is the common case and must be reported by name");
+    }
+
+    @Test
+    void everyKeyThisBuildUnderstandsIsAbsentFromTheUnknownList() {
+        assertTrue(BridgedConfig.unknownTopLevelKeys("""
+                bind:
+                  port: 8080
+                herdrSocket: /tmp/s
+                workers: {}
+                defaultWorker: a
+                guard: {}
+                worktreeRoot: /tmp
+                lifecycle: {}
+                spawnReadyTimeoutMs: 1
+                spawnReadyPollMs: 1
+                broker: {}
+                primary: {}
+                leaders: {}
+                leadScan: {}
+                placement: fixed
+                auth: {}
+                """).isEmpty(), "the known-key set must not drift from the record components");
+    }
+
+    @Test
+    void aMalformedOrEmptyDocumentIsNotReportedAsUnknownKeys() {
+        assertTrue(BridgedConfig.unknownTopLevelKeys("").isEmpty());
+        assertTrue(BridgedConfig.unknownTopLevelKeys("just a scalar").isEmpty());
+    }
+
+    // ── CB-531: lead discovery by tab label ─────────────────────────────────────────────────────
+
+    @Test
+    void leadScanIsOffUnlessTheBlockIsPresent(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-scan.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\n");
+
+        assertNull(BridgedConfig.load(f).leadScan(),
+                "turning this on widens who resolves as PRIMARY — upgrading the daemon must not do that");
+    }
+
+    @Test
+    void leadScanDefaultsItsFieldsWhenTheBlockIsPresentButBare(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("bare-scan.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\nleadScan: {}\n");
+
+        BridgedConfig.LeadScan scan = BridgedConfig.load(f).leadScan();
+        assertEquals("lead:", scan.tabPrefix());
+        assertEquals(10, scan.intervalSeconds());
+    }
+
+    @Test
+    void leadScanReadsAnExplicitPrefixAndInterval(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("scan.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                leadScan:
+                  tabPrefix: "drive:"
+                  intervalSeconds: 30
+                """);
+
+        BridgedConfig.LeadScan scan = BridgedConfig.load(f).leadScan();
+        assertEquals("drive:", scan.tabPrefix());
+        assertEquals(30, scan.intervalSeconds());
+    }
+
+    /**
+     * The hazard the guard exists for: bridged writes worker tab labels and reads lead tab labels.
+     * Overlap the two and every worker it spawns is read back as a lead.
+     */
+    @Test
+    void aLeadPrefixThatAWorkerTabLabelAlsoMatchesRefusesToStart(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("collide.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                workers:
+                  gx10:
+                    tabLabel: "lead: {profile} #{n}"
+                leadScan:
+                  tabPrefix: "lead:"
+                """);
+        BridgedConfig cfg = BridgedConfig.load(f);
+
+        IllegalStateException e = assertThrows(IllegalStateException.class, cfg::validateLeadScan);
+        assertTrue(e.getMessage().contains("gx10"), "the message must name the offending profile");
+    }
+
+    @Test
+    void theDefaultWorkerTabLabelDoesNotCollideWithTheDefaultLeadPrefix(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("ok.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                workers:
+                  gx10:
+                    baseUrl: http://gx00.gw:8000
+                leadScan: {}
+                """);
+
+        assertDoesNotThrow(() -> BridgedConfig.load(f).validateLeadScan());
+    }
+
+    @Test
+    void theCollisionGuardIsANoOpWhenScanningIsOff(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("off.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                workers:
+                  gx10:
+                    tabLabel: "lead: {profile}"
+                """);
+
+        assertDoesNotThrow(() -> BridgedConfig.load(f).validateLeadScan(),
+                "a label that collides with a convention nobody reads is not a problem");
+    }
+
+    // ── CB-530: the leaders registry ────────────────────────────────────────────────────────────
+
+    @Test
+    void leadersBlockRegistersEveryPaneByName(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("leaders.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                leaders:
+                  opus-5.0:
+                    terminal: term_opus
+                    kind: claude
+                  gpt-sol-5.6:
+                    terminal: term_sol
+                    kind: opencode
+                    model: openai/gpt-5.6-terra
+                """);
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+
+        assertEquals(Set.of("opus-5.0", "gpt-sol-5.6"), cfg.leaders().keySet());
+        assertEquals("opencode", cfg.leaders().get("gpt-sol-5.6").kind());
+        assertEquals("openai/gpt-5.6-terra", cfg.leaders().get("gpt-sol-5.6").model());
+        // The whole point: BOTH panes resolve as leads, so neither is demoted to worker.
+        assertEquals(Map.of("term_opus", "opus-5.0", "term_sol", "gpt-sol-5.6"),
+                cfg.leaderTerminals());
+    }
+
+    @Test
+    void aLegacyPrimaryPinAloneStillRegistersAsALeadNamedPrimary(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("legacy-pin.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\nprimary:\n  terminal: term_fixed\n");
+
+        assertEquals(Map.of("term_fixed", "primary"), BridgedConfig.load(f).leaderTerminals(),
+                "configs that never migrate must behave exactly as they did before CB-530");
+    }
+
+    @Test
+    void anExplicitLeadersEntryWinsOverThePinForTheSameTerminal(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("both.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                primary:
+                  terminal: term_shared
+                leaders:
+                  opus-5.0:
+                    terminal: term_shared
+                """);
+
+        assertEquals(Map.of("term_shared", "opus-5.0"), BridgedConfig.load(f).leaderTerminals(),
+                "the pin is the older spelling of the same fact; the named entry is what was meant");
+    }
+
+    @Test
+    void bothBlocksTogetherRegisterTheUnionOfTheirTerminals(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("union.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                primary:
+                  terminal: term_pinned
+                leaders:
+                  gpt-sol-5.6:
+                    terminal: term_sol
+                """);
+
+        assertEquals(Map.of("term_pinned", "primary", "term_sol", "gpt-sol-5.6"),
+                BridgedConfig.load(f).leaderTerminals());
+    }
+
+    @Test
+    void neitherBlockLeavesNothingRegistered(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("none.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\n");
+
+        assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty());
+    }
+
+    /** A lead entry with no terminal identifies nothing — it must not register a null key. */
+    @Test
+    void aLeadWithoutATerminalIsNotRegistered(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-terminal.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                leaders:
+                  sketch:
+                    kind: opencode
+                  real:
+                    terminal: term_real
+                """);
+
+        assertEquals(Map.of("term_real", "real"), BridgedConfig.load(f).leaderTerminals());
+    }
+
    @Test
    void absentBrokerBlockLeavesInboxSoftState(@TempDir Path dir) throws Exception {
        Path f = dir.resolve("no-broker.yaml");
@@ -345,6 +575,7 @@ class BridgedConfigTest {
                  idleTtlSeconds: 300
                  contextCap: 10
                  drainTimeoutSeconds: 5
+                  clearAfterTurn: true
                broker:
                  uri: amqp://guest:guest@127.0.0.1:5672
                primary:
@@ -371,6 +602,7 @@ class BridgedConfigTest {
        assertEquals(300, cfg.lifecycle().idleTtlSeconds());
        assertEquals(10, cfg.lifecycle().contextCap());
        assertEquals(5, cfg.lifecycle().drainTimeoutSeconds());
+        assertTrue(cfg.lifecycle().clearAfterTurn());
        assertEquals("amqp://guest:guest@127.0.0.1:5672", cfg.broker().uri());
        assertEquals("term_abc123", cfg.primary().terminal());
        assertEquals(5, cfg.primary().remindersOrDefault());
@@ -0,0 +1,262 @@
+package dev.ltms.bridged.herdr;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * CB-531. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that an
+ * operator-labelled tab is what makes a pane a lead, and — just as importantly — what does not.
+ */
+class LeadTabScannerTest {
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final long TTL = TimeUnit.SECONDS.toNanos(10);
+
+    /**
+     * A herdr whose workspace/tab/pane topology is declared per test. Counts calls so the caching
+     * contract can be asserted, and can be made to fail on demand.
+     */
+    private static final class TopologyHerdr implements HerdrClient {
+        /** workspace_id → label. */
+        final Map<String, String> workspaces = new LinkedHashMap<>();
+        /** tab_id → [workspace_id, label]. */
+        final Map<String, String[]> tabs = new LinkedHashMap<>();
+        /** pane_id → [tab_id, terminal_id]. */
+        final Map<String, String[]> panes = new LinkedHashMap<>();
+        int calls;
+        boolean failing;
+
+        TopologyHerdr workspace(String id, String label) {
+            workspaces.put(id, label);
+            return this;
+        }
+
+        TopologyHerdr tab(String tabId, String workspaceId, String label) {
+            tabs.put(tabId, new String[]{workspaceId, label});
+            return this;
+        }
+
+        TopologyHerdr pane(String paneId, String tabId, String terminalId) {
+            panes.put(paneId, new String[]{tabId, terminalId});
+            return this;
+        }
+
+        @Override
+        public JsonNode call(String method, Object params) {
+            calls++;
+            if (failing) {
+                throw new HerdrException("socket closed");
+            }
+            List<String> items = new ArrayList<>();
+            switch (method) {
+                case "workspace.list" -> {
+                    workspaces.forEach((id, label) -> items.add(
+                            "{\"workspace_id\":\"%s\",\"label\":\"%s\"}".formatted(id, label)));
+                    return read("{\"workspaces\":[%s]}".formatted(String.join(",", items)));
+                }
+                case "tab.list" -> {
+                    String ws = String.valueOf(((Map<?, ?>) params).get("workspace_id"));
+                    tabs.forEach((id, t) -> {
+                        if (ws.equals(t[0])) {
+                            items.add(("{\"tab_id\":\"%s\",\"workspace_id\":\"%s\",\"label\":%s,"
+                                    + "\"pane_count\":1}").formatted(id, t[0],
+                                    t[1] == null ? "null" : "\"" + t[1] + "\""));
+                        }
+                    });
+                    return read("{\"tabs\":[%s]}".formatted(String.join(",", items)));
+                }
+                case "pane.list" -> {
+                    panes.forEach((id, p) -> items.add(
+                            "{\"pane_id\":\"%s\",\"tab_id\":\"%s\",\"terminal_id\":\"%s\"}"
+                                    .formatted(id, p[0], p[1])));
+                    return read("{\"panes\":[%s]}".formatted(String.join(",", items)));
+                }
+                default -> throw new AssertionError("unexpected herdr call: " + method);
+            }
+        }
+
+        private static JsonNode read(String json) {
+            try {
+                return MAPPER.readTree(json);
+            } catch (Exception e) {
+                throw new AssertionError(e);
+            }
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
+    /**
+     * The usual shape: one user space with lead tabs, one worker space bridged owns.
+     *
+     * <p>Not closed: {@code close()} is a no-op on this fake, and every test needs the handle after
+     * the scanner is built (to mutate the topology or read {@code calls}).
+     */
+    @SuppressWarnings("resource")
+    private TopologyHerdr twoLeads() {
+        return new TopologyHerdr()
+                .workspace("w1", "main")
+                .workspace("w9", "bridged-workers")
+                .tab("w1:t1", "w1", "lead: opus-5.0")
+                .tab("w1:t2", "w1", "lead: gpt-sol-5.6")
+                .tab("w1:t3", "w1", "notes")
+                .tab("w9:t1", "w9", "worker: gx10 #1")
+                .pane("w1:p1", "w1:t1", "term_opus")
+                .pane("w1:p2", "w1:t2", "term_gpt")
+                .pane("w1:p3", "w1:t3", "term_notes")
+                .pane("w9:p1", "w9:t1", "term_worker");
+    }
+
+    private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> configured,
+                                   AtomicLong clock) {
+        return new LeadTabScanner(herdr, "lead:", Set.of("bridged-workers"), configured, TTL,
+                clock::get);
+    }
+
+    @Test
+    void everyLabelledTabBecomesALeadNamedByItsLabel() {
+        Map<String, String> leads = scanner(twoLeads(), Map.of(), new AtomicLong()).get();
+
+        assertEquals(Map.of("term_opus", "opus-5.0", "term_gpt", "gpt-sol-5.6"), leads,
+                "two leads discovered from labels alone — no terminal_id was ever configured");
+    }
+
+    @Test
+    void anUnlabelledTabContributesNothing() {
+        assertFalse(scanner(twoLeads(), Map.of(), new AtomicLong()).get().containsKey("term_notes"));
+    }
+
+    /**
+     * The guard that matters: bridged 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.
+     */
+    @Test
+    void aTabInAWorkerSpaceIsNeverALeadEvenWhenItsLabelMatches() {
+        TopologyHerdr herdr = twoLeads().tab("w9:t2", "w9", "lead: impostor")
+                .pane("w9:p2", "w9:t2", "term_impostor");
+
+        assertFalse(scanner(herdr, Map.of(), new AtomicLong()).get().containsKey("term_impostor"));
+    }
+
+    @Test
+    void aBarePrefixNamesNobodyAndIsRejected() {
+        TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
+                .tab("w1:t1", "w1", "lead:").pane("w1:p1", "w1:t1", "term_a");
+
+        assertEquals(Map.of(), scanner(herdr, Map.of(), new AtomicLong()).get(),
+                "a lead with no name would resolve as PRIMARY with nothing to attribute it to");
+    }
+
+    @Test
+    void thePrefixMatchesCaseInsensitivelyAndTheNameIsTrimmed() {
+        TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
+                .tab("w1:t1", "w1", "  LEAD:   opus-5.0  ").pane("w1:p1", "w1:t1", "term_a");
+
+        assertEquals(Map.of("term_a", "opus-5.0"), scanner(herdr, Map.of(), new AtomicLong()).get());
+    }
+
+    @Test
+    void everyPaneInALeadTabResolvesAsThatLead() {
+        // A human may split their own lead tab. Both panes are theirs, so both are that lead —
+        // nothing bridged placed can land here (see the worker-space test above).
+        TopologyHerdr herdr = twoLeads().pane("w1:p1b", "w1:t1", "term_opus_split");
+
+        assertEquals("opus-5.0", scanner(herdr, Map.of(), new AtomicLong()).get().get("term_opus_split"));
+    }
+
+    @Test
+    void anExplicitlyConfiguredLeadIsMergedInAndOutranksALabel() {
+        Map<String, String> configured = Map.of("term_opus", "pinned-name", "term_extra", "from-config");
+
+        Map<String, String> leads = scanner(twoLeads(), configured, new AtomicLong()).get();
+
+        assertEquals("pinned-name", leads.get("term_opus"), "an explicit pin is the operator's last word");
+        assertEquals("from-config", leads.get("term_extra"), "a configured lead needs no tab at all");
+        assertEquals("gpt-sol-5.6", leads.get("term_gpt"));
+    }
+
+    // ── caching ─────────────────────────────────────────────────────────────────────────────────
+
+    @Test
+    void aSecondLookupWithinTheTtlDoesNotTouchHerdr() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+
+        s.get();
+        int afterFirst = herdr.calls;
+        clock.addAndGet(TTL - 1);
+        s.get();
+
+        assertEquals(afterFirst, herdr.calls,
+                "resolve() runs on every request — an un-cached scan would put herdr on that path");
+    }
+
+    @Test
+    void aTabLabelledAfterStartupIsPickedUpOnceTheTtlExpires() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        assertFalse(s.get().containsKey("term_notes"));
+
+        herdr.tab("w1:t3", "w1", "lead: late-arrival"); // the operator renames their tab
+        clock.addAndGet(TTL);
+
+        assertEquals("late-arrival", s.get().get("term_notes"),
+                "the whole point over `leaders:`: no config edit, no restart");
+    }
+
+    @Test
+    void aFailedScanKeepsTheLeadsAlreadyKnownRatherThanDemotingThem() {
+        TopologyHerdr herdr = twoLeads();
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+        Map<String, String> before = s.get();
+
+        herdr.failing = true;
+        clock.addAndGet(TTL);
+
+        assertEquals(before, s.get(),
+                "a herdr hiccup must not silently demote a live lead to a worker mid-session");
+    }
+
+    @Test
+    void aFailedFirstScanStillHonoursTheConfiguredLeads() {
+        TopologyHerdr herdr = twoLeads();
+        herdr.failing = true;
+
+        Map<String, String> leads = scanner(herdr, Map.of("term_x", "opus-5.0"), new AtomicLong()).get();
+
+        assertEquals(Map.of("term_x", "opus-5.0"), leads,
+                "config-named leads must not depend on herdr answering at all");
+    }
+
+    @Test
+    void aDownHerdrIsRetriedOncePerTtlNotOncePerRequest() {
+        TopologyHerdr herdr = twoLeads();
+        herdr.failing = true;
+        AtomicLong clock = new AtomicLong();
+        LeadTabScanner s = scanner(herdr, Map.of(), clock);
+
+        s.get();
+        int afterFirst = herdr.calls;
+        s.get();
+        s.get();
+
+        assertEquals(afterFirst, herdr.calls, "the failure path must be rate-limited too");
+    }
+}
@@ -156,6 +156,21 @@ class CompletionResolverTest {
        assertEquals("No, 391 = 17 × 23.", waiter.getNow(null).text());
    }

+    @Test
+    void resolvesSynchronouslyBeforePostTurnContextClearing() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ previous answer\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
+        var waiter = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a");
+        herdr.readText("⏺ answer that /clear would erase\n❯ ");
+
+        resolver.resolveBeforePostAction("term_a");
+
+        assertTrue(waiter.isDone(), "the answer is captured before the adapter sends /clear");
+        assertEquals("answer that /clear would erase", waiter.getNow(null).text());
+    }
+
    @Test
    void suppressesAnUnchangedCompletionEvenWhenTheBlockExceedsTheScrapeCap() {
        // The fan-out issue-hunt finding: captureBaseline once stored the RAW (unclipped) assistant
@@ -196,6 +196,47 @@ class InjectorTest {
        assertEquals(List.of(T), completed, "a confirmed working→idle fires exactly one completion");
    }

+    @Test
+    void postTurnResetSettlesBeforeNextDelegationWithoutBecomingATurn() {
+        AgentControl agents = new AgentControl(herdr);
+        class ResetListener implements TurnListener {
+            int completed;
+
+            @Override
+            public void onTurnComplete(String target) {
+                completed++;
+            }
+
+            @Override
+            public boolean hasPostTurnAction(String target) {
+                return true;
+            }
+
+            @Override
+            public boolean onTurnCompleteWithPostAction(String target) {
+                completed++;
+                agents.send(target, "/clear"); // direct housekeeping, never Injector.enqueue
+                return true;
+            }
+        }
+        ResetListener listener = new ResetListener();
+        Injector inj = new Injector(agents, listener);
+        inj.enqueue(T, "first");
+        inj.enqueue(T, "second");
+
+        inj.onStatus(T, AgentStatus.IDLE);    // first delegation
+        inj.onStatus(T, AgentStatus.WORKING);
+        inj.onStatus(T, AgentStatus.IDLE);    // first complete; reset dispatched
+        assertEquals(List.of("first", "/clear"), sent(),
+                "same-tick completion must not let the queued delegation overtake reset");
+
+        inj.onStatus(T, AgentStatus.WORKING); // reset picked up, but this is not a bridge turn
+        inj.onStatus(T, AgentStatus.IDLE);    // reset settled; second may now deliver
+        assertEquals(List.of("first", "/clear", "second"), sent());
+        assertEquals(1, listener.completed,
+                "reset settlement must not recursively emit another turn completion");
+    }
+
    @Test
    void doesNotSynthesizeCompletionFromAnUnconfirmedTurn() {
        List<String> completed = new ArrayList<>();
@@ -288,7 +288,8 @@ class BridgeMcpTest {
        WorkerSession s = sessions.acquire("ltms-local", null, "/caller/proj", "term_primary",
                new WorktreeRequest("cb-304", null));

-        McpSchema.CallToolResult res = BridgeMcp.listWorkers(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions);
+        McpSchema.CallToolResult res = BridgeMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, Map.of(), "");

        assertNotEquals(Boolean.TRUE, res.isError());
        String out = textOf(res);
@@ -302,6 +303,58 @@ class BridgeMcpTest {
        assertTrue(out.contains("\"liveStatus\":\"unknown\""), out);
    }

+    @Test
+    void listReportsLeadsAndFlagsTheCallersOwnRow() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+
+        McpSchema.CallToolResult res = BridgeMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions,
+                Map.of("term_me", "opus-5.0", "term_peer", "gpt-sol-5.6"), "term_me");
+
+        assertNotEquals(Boolean.TRUE, res.isError());
+        String out = textOf(res);
+        assertTrue(out.contains("\"name\":\"opus-5.0\""), out);
+        assertTrue(out.contains("\"name\":\"gpt-sol-5.6\""), out);
+        assertTrue(out.contains("\"sessionId\":\"term_peer\""), out);
+        // The caller's own row is flagged, and only the caller's — a peer must be distinguishable
+        // from self without a second bridge_whoami call.
+        assertEquals(1, out.split("\"self\":true", -1).length - 1, out);
+        assertTrue(out.indexOf("term_me") < out.indexOf("\"self\":true"), out);
+    }
+
+    @Test
+    void listReportsBothHalvesEvenWhenEmpty() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+
+        McpSchema.CallToolResult res = BridgeMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, Map.of(), "");
+
+        // An absent "leads" key is what made an empty worker roster read as "no peers" (CB-535).
+        String out = textOf(res);
+        assertTrue(out.contains("\"leads\":[]"), out);
+        assertTrue(out.contains("\"workers\":[]"), out);
+    }
+
+    @Test
+    void listReportsALeadHerdrCannotSeeAsUnknown() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+
+        McpSchema.CallToolResult res = BridgeMcp.listFleet(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions,
+                Map.of("term_ghost", "gone-away"), "term_me");
+
+        // Reported, not hidden: an unreachable peer is exactly what a would-be sender needs to see.
+        String out = textOf(res);
+        assertTrue(out.contains("\"name\":\"gone-away\""), out);
+        assertTrue(out.contains("\"status\":\"unknown\""), out);
+    }
+
    @Test
    void stopTearsDownAWorkerByPane() {
        FakeHerdr h = new FakeHerdr();
@@ -105,4 +105,68 @@ class PrimaryRegistryTest {
        reg.record("term_found");
        assertEquals("term_found", reg.primaryTerminal().get());
    }
+
+    // ── CB-532: nudges follow the delegating lead, not "the primary" ────────────────────────────
+
+    /**
+     * The bug that made `primary.terminal` unretirable: with one slot, whichever lead called
+     * bridge_send first captured every nudge — including nudges for the other lead's delegations.
+     */
+    @Test
+    void aNudgeGoesToTheLeadThatDelegatedToThatWorker() {
+        var reg = new PrimaryRegistry(null);
+        reg.recordDelegation("term_worker_a", "term_lead_opus");
+        reg.recordDelegation("term_worker_b", "term_lead_sol");
+
+        assertEquals("term_lead_opus", reg.nudgeTargetFor("term_worker_a").orElseThrow());
+        assertEquals("term_lead_sol", reg.nudgeTargetFor("term_worker_b").orElseThrow());
+    }
+
+    @Test
+    void theMostRecentDelegatorWinsWhenAWorkerChangesHands() {
+        var reg = new PrimaryRegistry(null);
+        reg.recordDelegation("term_worker", "term_lead_opus");
+        reg.recordDelegation("term_worker", "term_lead_sol");
+
+        assertEquals("term_lead_sol", reg.nudgeTargetFor("term_worker").orElseThrow(),
+                "the lead waiting on the reply is the one that sent the work most recently");
+    }
+
+    /** After a restart the map is empty while the durable inbox still holds the reply. */
+    @Test
+    void anUnknownDelegationFallsBackToThePinnedPrimary() {
+        var reg = new PrimaryRegistry("term_pinned");
+
+        assertEquals("term_pinned", reg.nudgeTargetFor("term_never_seen").orElseThrow());
+    }
+
+    @Test
+    void withNoPinAndNoDelegationNobodyIsNudged() {
+        var reg = new PrimaryRegistry(null);
+
+        assertTrue(reg.nudgeTargetFor("term_worker").isEmpty(),
+                "guessing would interrupt the wrong lead with someone else's result; the durable "
+                        + "inbox makes pull the correct degradation");
+    }
+
+    @Test
+    void releasingASessionForgetsItsLead() {
+        var reg = new PrimaryRegistry(null);
+        reg.recordDelegation("term_worker", "term_lead");
+
+        reg.forgetDelegation("term_worker");
+
+        assertTrue(reg.nudgeTargetFor("term_worker").isEmpty());
+    }
+
+    @Test
+    void aBlankOrNullDelegationIsIgnoredRatherThanStored() {
+        var reg = new PrimaryRegistry(null);
+        reg.recordDelegation("term_worker", "  ");
+        reg.recordDelegation(null, "term_lead");
+        reg.recordDelegation("  ", "term_lead");
+
+        assertTrue(reg.nudgeTargetFor("term_worker").isEmpty());
+        assertTrue(reg.nudgeTargetFor(null).isEmpty());
+    }
 }
@@ -39,13 +39,18 @@ class SessionManagerTest {
    }

    private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap) {
+        return sessionManager(herdr, clock, contextCap, false);
+    }
+
+    private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
+                                          boolean clearAfterTurn) {
        BridgedConfig.Worker cfg = new BridgedConfig.Worker(
                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN",
                List.of("ccs", "ltms-local"), "tab", "bridged-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);
-        return new SessionManager(workers, new GitWorktrees(), clock, contextCap);
+        return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn);
    }

    @Test
@@ -336,6 +341,50 @@ class SessionManagerTest {
                "forced release tears the worker pane down exactly once");
    }

+    @Test
+    void clearAfterTurnResetsContextWithoutDoubleCountingTheTurn() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr, () -> 0L, 0, true);
+        WorkerSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+
+        sessions.onDelivered(session.terminalId());
+        assertTrue(sessions.onTurnCompleteWithPostAction(session.terminalId()));
+
+        WorkerSession updated = sessions.get(session.paneId()).orElseThrow();
+        assertEquals(1, updated.turnCount(), "the reset is housekeeping, not a second delegation");
+        assertEquals(List.of("/clear"), promptTexts(herdr));
+    }
+
+    @Test
+    void contextCapReleaseWinsOverClearAfterTurn() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr, () -> 0L, 1, true);
+        WorkerSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId());
+
+        assertFalse(sessions.hasPostTurnAction(session.terminalId()),
+                "a session at its cap will be released, not reset for reuse");
+        assertFalse(sessions.onTurnCompleteWithPostAction(session.terminalId()));
+        assertTrue(sessions.get(session.paneId()).isEmpty());
+        assertTrue(promptTexts(herdr).isEmpty(), "never send /clear into a worker being torn down");
+    }
+
+    @Test
+    void clearAfterTurnFalsePreservesCompletionWithoutAControlPrompt() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr, () -> 0L, 0, false);
+        WorkerSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId());
+
+        sessions.onTurnComplete(session.terminalId());
+
+        assertEquals(WorkerSession.State.DONE, sessions.get(session.paneId()).orElseThrow().state());
+        assertTrue(promptTexts(herdr).isEmpty());
+    }
+
    @Test
    void drainAllReleasesBusyAndReadySessionsAndWaitsForBusy() {
        long[] clock = {0};
@@ -367,6 +416,13 @@ class SessionManagerTest {
                .count();
    }

+    private static List<String> promptTexts(FakeHerdr herdr) {
+        return herdr.calls.stream()
+                .filter(c -> "agent.prompt".equals(c.method()))
+                .map(c -> String.valueOf(((Map<?, ?>) c.params()).get("text")))
+                .toList();
+    }
+
    // --- CB-306 spawn-readiness gate: no half-registered session on timeout ----------------

    @Test
@@ -78,14 +78,24 @@ class ClaudeCodeLauncherTest {

        Map<?, ?> start = (Map<?, ?>) herdr.lastCall("agent.start").params();
        assertEquals("claude", start.get("kind"), "herdr launches the canonical executable by kind");
-        assertEquals(List.of(), start.get("args"), "the configured executable is not repeated in args");
+        // CB-533: the shared fixture pins model "coder", so the model flag is the whole args list.
+        // What this test guards is that argv[0] is NOT repeated — herdr supplies it from `kind`.
+        assertEquals(List.of("--model", "coder"), start.get("args"),
+                "the configured executable is not repeated in args");
    }

    @Test
    void noBridgeFlagsWhenMcpUrlAbsent() {
        FakeHerdr herdr = new FakeHerdr();
        service(herdr, List.of("claude", "--verbose"), null).spawn();
-        assertEquals(List.of("--verbose"), spawnedArgs(herdr), "extra args untouched without mcpUrl");
+
+        List<String> args = spawnedArgs(herdr);
+        assertFalse(args.contains("--mcp-config"), "no bridge mount without mcpUrl");
+        assertFalse(args.contains("--append-system-prompt"), "no reply charter without mcpUrl");
+        // CB-533: the model flag is independent of the MCP mount — pinning the model is not part of
+        // "mount the bridge", so an unmounted worker still runs the model its profile names.
+        assertEquals(List.of("--verbose", "--model", "coder"), args,
+                "the operator's own args are preserved, in order, ahead of the model flag");
    }

    private ClaudeCodeLauncher multiProfile(FakeHerdr herdr) {
@@ -293,6 +303,20 @@ class ClaudeCodeLauncherTest {
        assertTrue(caps.contains(Capability.MID_TURN_ASK), "every Claude Code peer supports mid-turn ask");
        assertTrue(caps.contains(Capability.WORKTREE), "every CLI peer supports worktree cwd");
        assertTrue(caps.contains(Capability.ORPHAN_REAP), "every herdr launcher supports orphan reap");
+        assertTrue(caps.contains(Capability.CONTEXT_RESET), "Claude Code supports /clear");
+    }
+
+    @Test
+    void clearContextUsesTheClaudeCommandThroughTheOwningHandle() {
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher svc = service(herdr, List.of("claude"), null);
+        PeerHandle handle = svc.spawn(new SpawnRequest(null, null, null));
+
+        assertTrue(svc.clearContext(handle.id()));
+
+        Map<?, ?> prompt = (Map<?, ?>) herdr.lastCall("agent.prompt").params();
+        assertEquals("/clear", prompt.get("text"));
+        assertEquals("w9:pRoot_1", prompt.get("target"));
    }

    @Test
@@ -541,4 +565,54 @@ class ClaudeCodeLauncherTest {
        assertEquals("http://gx00.gw:8000", startEnv(herdr).get("ANTHROPIC_BASE_URL"),
                "the guard-checked baseUrl must win over any env: entry, or the boundary is bypassable");
    }
+
+    // ── CB-533: the model is pinned on the command line, not only in the environment ────────────
+
+    /** A launcher for a profile identical but for its {@code model:} — the only variable here. */
+    private ClaudeCodeLauncher serviceWithModel(FakeHerdr herdr, String model) {
+        BridgedConfig.Worker cfg = profileWithModel(model);
+        return new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
+                _ -> null);
+    }
+
+    private static BridgedConfig.Worker profileWithModel(String model) {
+        return new BridgedConfig.Worker("sonnet", "http://gx00.gw:8000", model, null,
+                "BRIDGED_WORKER_TOKEN", List.of("ccs", "sonnet"), "tab", "bridged-workers",
+                "w #{n}", "http://127.0.0.1:8765/mcp", null, null);
+    }
+
+    @Test
+    void aConfiguredModelIsPassedAsAModelFlagAsWellAsTheEnvVar() {
+        // ANTHROPIC_MODEL alone loses to `ccs`, which exports its own model family over whatever it
+        // inherited — so a profile that set model: was silently overruled by its own launcher.
+        FakeHerdr herdr = new FakeHerdr();
+        serviceWithModel(herdr, "claude-sonnet-5").spawn("sonnet", null, null);
+
+        assertEquals("claude-sonnet-5", startEnv(herdr).get("ANTHROPIC_MODEL"));
+        List<String> args = spawnedArgs(herdr);
+        int flag = args.indexOf("--model");
+        assertTrue(flag >= 0, "the flag is what survives a wrapper argv like [ccs, sonnet]");
+        assertEquals("claude-sonnet-5", args.get(flag + 1));
+    }
+
+    @Test
+    void theModelFlagComesLastSoItOutranksTheOperatorsOwnArgv() {
+        FakeHerdr herdr = new FakeHerdr();
+        serviceWithModel(herdr, "claude-sonnet-5").spawn("sonnet", null, null);
+
+        List<String> args = spawnedArgs(herdr);
+        assertEquals(args.size() - 2, args.indexOf("--model"));
+    }
+
+    @Test
+    void aProfileWithNoModelGetsNoModelFlag() {
+        // gx10 deliberately leaves model: unset so ccs owns selection; adding a flag would make
+        // this file a second source of truth for exactly the thing it declines to decide.
+        FakeHerdr herdr = new FakeHerdr();
+        serviceWithModel(herdr, null).spawn("sonnet", null, null);
+
+        assertFalse(spawnedArgs(herdr).contains("--model"));
+        assertNull(startEnv(herdr).get("ANTHROPIC_MODEL"));
+    }
 }
@@ -1,5 +1,8 @@
 package dev.ltms.bridged.worker;

+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import dev.ltms.bridged.config.BridgedConfig;
 import dev.ltms.bridged.guard.SubscriptionGuard;
 import dev.ltms.bridged.herdr.Agent;
@@ -14,6 +17,7 @@ import dev.ltms.bridged.peer.SpawnRequest;
 import dev.ltms.bridged.placement.PlacementException;
 import dev.ltms.bridged.placement.PlacementPolicies;
 import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;

 import java.util.EnumSet;
 import java.util.HashMap;
@@ -232,6 +236,30 @@ class CompositePeerLauncherTest {
                "stop routes to the spawning adapter and closes exactly that worker's pane");
    }

+    @Test
+    void opencodeContextResetIsANoOpAndWarnsOnlyOnce() {
+        FakeHerdr herdr = new FakeHerdr();
+        CompositePeerLauncher composite = composite(herdr);
+        PeerHandle handle = composite.spawn(new SpawnRequest("gemini", null, null));
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            assertFalse(composite.clearContext(handle.id()));
+            assertFalse(composite.clearContext(handle.id()));
+        } finally {
+            logger.detachAppender(appender);
+        }
+
+        assertFalse(opencodeAdapter(herdr).capabilities().contains(Capability.CONTEXT_RESET));
+        assertTrue(herdr.calls.stream().noneMatch(c -> "agent.prompt".equals(c.method())),
+                "never type Claude's /clear into an opencode prompt");
+        assertEquals(1, appender.list.stream()
+                .filter(e -> e.getFormattedMessage().contains("context reset is unsupported"))
+                .count(), "unsupported reset is logged once per adapter, not once per turn");
+    }
+
    @Test
    void constructorRejectsAProfileClaimedByTwoAdapters() {
        FakeHerdr herdr = new FakeHerdr();