CB-578 stage B: quarantine the exhausted credential, not the profile

A BACKEND_EXHAUSTED classification (stage A) now puts that profile's
credential into a BackendQuarantine for a configurable cooldown. A spawn
onto a quarantined profile is refused naming the credential and roughly
when it lifts; weighted/round-robin/fixed placement skip a quarantined
candidate; the quarantine lifts itself on the injected clock; and it is
visible on bridge_profiles.

Keyed by credential, not by profile name, via the new Profile.credentialId
(profiles sharing one credential quarantine together — e.g. two models on
one account) and effectiveCredentialId() (unset ⇒ quarantines alone,
today's behaviour unchanged). Fixed a related gap along the way: a reload
changing exhaustedPattern was silently reported "applied" even though it's
deferred — sameLaunchSettings() now catches it too.

bridged/src/test/java/dev/ltms/bridged/config/ConfigRefTest.java

@@ -338,6 +338,54 @@ class ConfigRefTest {
         assertEquals("deepseek-v4-flash", ref.get().profiles().get("sonnet").model());
     }
 
+    /**
+     * CB-578 stage B: exhaustedPattern is compiled once into Bridged.main's pattern map at startup
+     * (see ExhaustedPatternLookup), so a reload never re-reads it — a changed pattern must be
+     * reported deferred exactly like model/baseUrl, not silently claimed as applied.
+     */
+    @Test
+    void changingAProfilesExhaustedPatternIsReportedAsDeferred(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("bridged.yaml");
+        Files.writeString(f, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: ~/.config/herdr/herdr.sock
+                profiles:
+                  sonnet:
+                    baseUrl: http://gx00.gw:8000
+                    model: sonnet
+                    exhaustedPattern: "usage limit has been reached"
+                guard:
+                  offSubscriptionHosts:
+                    - gx00.gw
+                """);
+        ConfigRef ref = refFor(f);
+
+        Files.writeString(f, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                herdrSocket: ~/.config/herdr/herdr.sock
+                profiles:
+                  sonnet:
+                    baseUrl: http://gx00.gw:8000
+                    model: sonnet
+                    exhaustedPattern: "rate limit exceeded"
+                guard:
+                  offSubscriptionHosts:
+                    - gx00.gw
+                """);
+        ConfigRef.Outcome out = ref.reload();
+
+        assertTrue(out.applied());
+        assertEquals(1, out.deferred().size(), out.deferred().toString());
+        assertTrue(out.deferred().getFirst().contains("sonnet"), out.deferred().toString());
+        assertTrue(out.deferred().getFirst().contains("launch settings"), out.deferred().toString());
+        // The snapshot still carries the new value — a restart is what makes it take effect.
+        assertEquals("rate limit exceeded", ref.get().profiles().get("sonnet").exhaustedPattern());
+    }
+
     @Test
     void aFixedRefHasNoFileAndRefusesToReload() {
         BridgedConfig cfg = new BridgedConfig(null, null, null, null, null, null,

bridged/src/test/java/dev/ltms/bridged/inject/CompletionResolverTest.java

@@ -26,7 +26,7 @@ class CompletionResolverTest {
     void skipsTheScrapeWhenNoSendIsWaiting() {
         FakeHerdr herdr = new FakeHerdr();
         Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         resolver.resolve("term_a", null); // no in-flight turn captured for this target
 
@@ -38,7 +38,7 @@ class CompletionResolverTest {
     void failSkipsTheScrapeWhenNoSendIsWaiting() {
         FakeHerdr herdr = new FakeHerdr();
         Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         resolver.fail("term_a", null); // no in-flight turn, and no registered waiter to fall back to
 
@@ -50,7 +50,7 @@ class CompletionResolverTest {
     void captureBaselineSkipsTheReadWhenNoSendIsWaiting() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ X\n❯ ");
         Rendezvous rendezvous = new Rendezvous(); // no waiter opened
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         resolver.captureBaseline("term_a", TestTurnTokens.inert("term_a")); // no send to attribute a later completion to
 
@@ -139,7 +139,7 @@ class CompletionResolverTest {
         // send must NOT be resolved with the stale answer.
         FakeHerdr herdr = new FakeHerdr().readText("⏺ 391\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");   // a send is blocked on this turn
         // The turn as captured at delivery: its waiter, and the previous turn's answer still on screen.
@@ -154,7 +154,7 @@ class CompletionResolverTest {
     void resolvesACompletionWhoseScrapeChangedSinceDelivery() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ No, 391 = 17 × 23.\n❯ "); // the worker's real answer
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         // Delivery baseline was the previous turn's "391"; the scrape now differs → resolve.
@@ -171,7 +171,7 @@ class CompletionResolverTest {
         String block = "⏺ " + "x".repeat(CompletionResolver.MAX_SCRAPE_CHARS + 1) + "\n❯ ";
         FakeHerdr herdr = new FakeHerdr().readText(block);
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -185,7 +185,7 @@ class CompletionResolverTest {
     void leavesAnUnclippedCompletionPaneTailUnmarked() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ complete report\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -197,7 +197,7 @@ class CompletionResolverTest {
     void resolvesSynchronouslyBeforePostTurnContextClearing() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ previous answer\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
         var waiter = rendezvous.open("term_a");
         resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
         herdr.readText("⏺ answer that /clear would erase\n❯ ");
@@ -219,7 +219,7 @@ class CompletionResolverTest {
         String longBlock = "⏺ " + "x".repeat(CompletionResolver.MAX_SCRAPE_CHARS + 500) + "\n❯ ";
         FakeHerdr herdr = new FakeHerdr().readText(longBlock);
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");        // a send is blocked on this turn
         resolver.captureBaseline("term_a", new TurnToken("term_a", waiter)); // baseline is the clipped >cap block
@@ -239,7 +239,7 @@ class CompletionResolverTest {
         // No delivery baseline (e.g. the pre-turn read failed) ⇒ never suppress; the completion resolves.
         FakeHerdr herdr = new FakeHerdr().readText("⏺ hello\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -257,7 +257,7 @@ class CompletionResolverTest {
         // byte-identical guard would wrongly match the empty tail and suppress.
         FakeHerdr herdr = new FakeHerdr().healthy(false); // agent.read throws HerdrException
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         var turn = new CompletionResolver.InFlight(waiter, ""); // empty pane baselined at delivery
@@ -277,7 +277,7 @@ class CompletionResolverTest {
         // fail must not overwrite that value, and must not even scrape the worker — nobody needs it.
         FakeHerdr herdr = new FakeHerdr().readText("an error screen");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         var turn = new CompletionResolver.InFlight(waiter, null);
@@ -298,7 +298,7 @@ class CompletionResolverTest {
         // fail falls back to the waiter currently registered on the Rendezvous and fails it.
         FakeHerdr herdr = new FakeHerdr().readText("stuck on an error screen");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a"); // send registered, but no captureBaseline ever ran
         resolver.fail("term_a", null);          // no in-flight turn → fall back to the registered waiter
@@ -324,7 +324,7 @@ class CompletionResolverTest {
         try {
             FakeHerdr herdr = new FakeHerdr().readText("stuck on an error screen");
             Rendezvous rendezvous = new Rendezvous();
-            CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+            CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
             var waiter = rendezvous.open("term_a");
 
             resolver.fail("term_a", null);
@@ -352,7 +352,7 @@ class CompletionResolverTest {
         // scrape to turn N+1; targeting turn N's captured waiter makes the late completion a no-op.
         FakeHerdr herdr = new FakeHerdr().readText("⏺ turn N answer\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiterN = rendezvous.open("term_a"); // turn N's send
         // The turn as the injector captured it at delivery (waiter + pre-turn baseline).
@@ -382,7 +382,7 @@ class CompletionResolverTest {
         FakeHerdr herdr = new FakeHerdr().readText(block);
         Rendezvous rendezvous = new Rendezvous();
         ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -398,7 +398,7 @@ class CompletionResolverTest {
         FakeHerdr herdr = new FakeHerdr().readText(block);
         Rendezvous rendezvous = new Rendezvous();
         ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -407,12 +407,53 @@ class CompletionResolverTest {
                 waiter.getNow(null).text(), "the reason names the real cause and carries the matched line");
     }
 
+    @Test
+    void aWinningBackendExhaustedClassificationNotifiesTheExhaustionSink() {
+        String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertEquals(1, notified.size(), "the sink is notified exactly once for the winning classification");
+        assertTrue(notified.get(0).startsWith("term_a: "), "the sink is told which target exhausted");
+        assertTrue(notified.get(0).contains("The usage limit has been reached"),
+                "the sink is told the matched reason: " + notified.get(0));
+    }
+
+    @Test
+    void aLosingBackendExhaustedClassificationNeverNotifiesTheExhaustionSink() {
+        // The waiter was already resolved (e.g. by the worker's own reply) before this scrape landed —
+        // resolveExhausted loses the race and must return false, so the sink must not fire either.
+        String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
+        java.util.List<String> notified = new java.util.ArrayList<>();
+        ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
+
+        var waiter = rendezvous.open("term_a");
+        var turn = new CompletionResolver.InFlight(waiter, null);
+        assertTrue(rendezvous.resolveCompletion(waiter, "already replied"));
+
+        resolver.resolve("term_a", turn);
+
+        assertTrue(notified.isEmpty(), "a classification that loses the race must not quarantine anything");
+        assertEquals("already replied", waiter.getNow(null).text(), "the earlier resolution stands untouched");
+    }
+
     @Test
     void aNonMatchingScrapeResolvesAsAnOrdinaryCompletion() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ complete report\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
         ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -430,7 +471,7 @@ class CompletionResolverTest {
         FakeHerdr herdr = new FakeHerdr().readText(block);
         Rendezvous rendezvous = new Rendezvous();
         CompletionResolver resolver =
-                new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
+                new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
 
         var waiter = rendezvous.open("term_a");
         resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java

@@ -72,7 +72,8 @@ class BridgeMcpAuthzTest {
                 new PrimaryRegistry(null),
                 enforce ? CallerResolver.withLeadsAndMembers(identity, false, null,
                 Map::of, new MemberRegistry(null)) : null,
-                metrics, BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"));
+                metrics, BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"),
+                BridgeMcp.QuarantineSource.none());
         return mcp;
     }
 

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java

@@ -16,6 +16,7 @@ import dev.ltms.bridged.inject.MemberPresence;
 import dev.ltms.bridged.session.MemberSession;
 import dev.ltms.bridged.session.WorktreeRequest;
 import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.placement.BackendQuarantine;
 import io.modelcontextprotocol.spec.McpSchema;
 import dev.ltms.bridged.msg.InMemoryReplyInbox;
 import org.junit.jupiter.api.BeforeEach;
@@ -437,11 +438,29 @@ class BridgeMcpTest {
     @Test
     void profilesListsConfiguredProfilesAndDefault() {
         FakeHerdr h = new FakeHerdr();
-        McpSchema.CallToolResult res = BridgeMcp.profiles(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        McpSchema.CallToolResult res = BridgeMcp.profiles(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), BridgeMcp.QuarantineSource.none());
         assertNotEquals(Boolean.TRUE, res.isError());
         String out = textOf(res);
         assertTrue(out.contains("ltms-local"), out);
         assertTrue(out.contains("\"default\":\"ltms-local\""), out);
+        assertFalse(out.contains("quarantined"), "no profile is quarantined, so the key is omitted: " + out);
+    }
+
+    @Test
+    void profilesReportsAQuarantinedCredential() {
+        FakeHerdr h = new FakeHerdr();
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        BridgeMcp.QuarantineSource source = new BridgeMcp.QuarantineSource(
+                profile -> "ltms-local".equals(profile) ? "shared-openai" : null, quarantine);
+        McpSchema.CallToolResult res = BridgeMcp.profiles(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), source);
+        assertNotEquals(Boolean.TRUE, res.isError());
+        String out = textOf(res);
+        assertTrue(out.contains("\"quarantined\""), out);
+        assertTrue(out.contains("shared-openai"), out);
+        assertTrue(out.contains("\"quarantinedForSeconds\":1800"), out);
     }
 
     @Test

bridged/src/test/java/dev/ltms/bridged/member/CompositePeerLauncherTest.java

@@ -16,6 +16,7 @@ import dev.ltms.bridged.peer.PeerHandle;
 import dev.ltms.bridged.peer.PeerLauncher;
 import dev.ltms.bridged.peer.PeerUnreachableException;
 import dev.ltms.bridged.peer.SpawnRequest;
+import dev.ltms.bridged.placement.BackendQuarantine;
 import dev.ltms.bridged.placement.PlacementException;
 import dev.ltms.bridged.placement.PlacementPolicies;
 import org.junit.jupiter.api.Test;
@@ -27,6 +28,8 @@ 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.*;
 
@@ -128,6 +131,13 @@ class CompositePeerLauncherTest {
                 weight, maxLoad);
     }
 
+    private static BridgedConfig.Profile stubWorker(String profile, String credentialId) {
+        return new BridgedConfig.Profile(profile, "http://gx00.gw:8000", "coder",
+                null, "BRIDGED_WORKER_TOKEN", List.of("claude"), "tab", "bridged-workers",
+                "w #{n}", null, null, null, null, null, null, null, null, null,
+                null, null, credentialId);
+    }
+
     /**
      * An <em>order-preserving</em> profile map. Never {@code Map.of} here: its iteration order is
      * salted per JVM run, and the weighted policy breaks an exact-weight tie on candidate order —
@@ -554,4 +564,105 @@ class CompositePeerLauncherTest {
                 new SpawnRequest(null, null, null, null, null, MemberRole.DEV)));
         assertTrue(e.getMessage().contains("maxLoad"), e.getMessage());
     }
+
+    // ── CB-578 stage B: a BACKEND_EXHAUSTED classification quarantines the credential ──────────
+
+    @Test
+    void explicitSpawnOntoAQuarantinedProfileIsRefusedNamingTheCredentialAndRemainingTime() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, BridgedConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "terra", stubWorker("terra", "shared-openai"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "sol", Set.of());
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        PlacementException e = assertThrows(PlacementException.class,
+                () -> composite.spawn(new SpawnRequest("sol", null, null)));
+        assertTrue(e.getMessage().contains("sol"), "message names the profile: " + e.getMessage());
+        assertTrue(e.getMessage().contains("shared-openai"), "message names the credential: " + e.getMessage());
+        assertTrue(e.getMessage().contains("1800"), "message names roughly when it lifts: " + e.getMessage());
+        assertEquals(0, adapter.spawnCount("sol"), "the quarantined profile is never delegated to");
+    }
+
+    /**
+     * The part CB-578 stage B calls out as easy to get wrong: sol and terra are two different
+     * profiles sharing one OpenAI credential. Quarantining because of an exhaustion classified on
+     * ONE of them must lock out the other too, or the fleet just walks onto the same dead account
+     * under the sibling's name.
+     */
+    @Test
+    void twoProfilesSharingACredentialAreBothQuarantinedByOneExhaustionEvent() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, BridgedConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "terra", stubWorker("terra", "shared-openai"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "sol", Set.of());
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        // Only "sol" was classified BACKEND_EXHAUSTED — but the two profiles share one credential.
+        quarantine.quarantine("shared-openai");
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        assertThrows(PlacementException.class, () -> composite.spawn(new SpawnRequest("sol", null, null)),
+                "sol was the one classified exhausted");
+        assertThrows(PlacementException.class, () -> composite.spawn(new SpawnRequest("terra", null, null)),
+                "terra shares sol's credential, so it must be locked out too");
+        assertEquals(0, adapter.spawnCount("sol"));
+        assertEquals(0, adapter.spawnCount("terra"));
+    }
+
+    @Test
+    void placementSkipsAQuarantinedProfileAndRoutesToAnUnquarantinedOne() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, BridgedConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "b", stubWorker("b"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "sol", Set.of());
+        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.weighted(), _ -> 0, null, quarantine);
+
+        PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
+        assertEquals("b", h.profile(), "sol is quarantined, so an unqualified spawn must land on b");
+        assertEquals(0, adapter.spawnCount("sol"));
+    }
+
+    @Test
+    void aQuarantineLiftsOnTheInjectedClockAndTheProfileBecomesSpawnableAgain() {
+        FakeHerdr herdr = new FakeHerdr();
+        Map<String, BridgedConfig.Profile> profiles = ordered(
+                "sol", stubWorker("sol", "shared-openai"),
+                "b", stubWorker("b"));
+        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "sol", Set.of());
+        AtomicLong nowNanos = new AtomicLong(0L);
+        BackendQuarantine quarantine = new BackendQuarantine(nowNanos::get, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai");
+        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
+                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
+
+        assertThrows(PlacementException.class, () -> composite.spawn(new SpawnRequest("sol", null, null)),
+                "still inside the cooldown");
+
+        nowNanos.set(TimeUnit.MINUTES.toNanos(31));
+
+        PeerHandle h = composite.spawn(new SpawnRequest("sol", null, null));
+        assertEquals("sol", h.profile(), "the cooldown expired on the injected clock — sol is spawnable again");
+        assertEquals(1, adapter.spawnCount("sol"));
+    }
+
+    @Test
+    void aFleetWithNoExhaustedPatternAnywhereBehavesExactlyAsBeforeQuarantineExisted() {
+        // BackendQuarantine.none() is the inert stand-in every constructor already defaults to when
+        // no quarantine is wired — the 2/5/6-arg constructors used throughout this file all exercise
+        // it. This test pins that an explicit .none() also never refuses a spawn, for any profile.
+        FakeHerdr herdr = new FakeHerdr();
+        PeerLauncher composite = composite(herdr);
+
+        assertDoesNotThrow(() -> composite.spawn(new SpawnRequest("claude", null, null)));
+        assertDoesNotThrow(() -> composite.spawn(new SpawnRequest("gemini", null, null)));
+    }
 }

bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java

@@ -6,6 +6,7 @@ import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.inject.CompletionResolver;
 import dev.ltms.bridged.inject.ExhaustedPatternLookup;
+import dev.ltms.bridged.inject.ExhaustionSink;
 import dev.ltms.bridged.mcp.PrimaryRegistry;
 import dev.ltms.bridged.inject.Injector;
 import org.junit.jupiter.api.BeforeEach;
@@ -33,7 +34,7 @@ class MessageServiceTest {
     private final AgentControl agents = new AgentControl(herdr);
     private final Rendezvous rendezvous = new Rendezvous();
     private final CompletionResolver completion =
-            new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none());
+            new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
     private final Injector injector = new Injector(agents, completion);
     private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
     private final MessageService messages = new MessageService(agents, injector, rendezvous, inbox);

bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java

@@ -0,0 +1,108 @@
+package dev.ltms.bridged.placement;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CB-578 stage B: the credential-keyed quarantine tracker itself, isolated from placement/spawn
+ * wiring (that's {@code CompositePeerLauncherTest}). The clock is a plain {@link AtomicLong} of
+ * nanos so expiry is exercised without a real sleep.
+ */
+class BackendQuarantineTest {
+
+    @Test
+    void aFreshCredentialIsNotQuarantined() {
+        BackendQuarantine q = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        assertFalse(q.isQuarantined("shared-openai"));
+        assertEquals(OptionalLong.empty(), q.remainingSeconds("shared-openai"));
+    }
+
+    @Test
+    void quarantineBlocksTheCredentialForTheFullCooldown() {
+        BackendQuarantine q = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        q.quarantine("shared-openai");
+
+        assertTrue(q.isQuarantined("shared-openai"));
+        assertEquals(OptionalLong.of(1800L), q.remainingSeconds("shared-openai"));
+    }
+
+    @Test
+    void onlyTheQuarantinedCredentialIsAffected() {
+        BackendQuarantine q = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
+        q.quarantine("shared-openai");
+
+        assertFalse(q.isQuarantined("some-other-credential"),
+                "an unrelated credential must not be swept into the quarantine");
+    }
+
+    @Test
+    void expiresOnTheInjectedClock() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.MINUTES.toNanos(30));
+        q.quarantine("shared-openai");
+        assertTrue(q.isQuarantined("shared-openai"));
+
+        now.set(TimeUnit.MINUTES.toNanos(29));
+        assertTrue(q.isQuarantined("shared-openai"), "still inside the cooldown");
+
+        now.set(TimeUnit.MINUTES.toNanos(31));
+        assertFalse(q.isQuarantined("shared-openai"), "the cooldown has elapsed on the injected clock");
+        assertEquals(OptionalLong.empty(), q.remainingSeconds("shared-openai"));
+    }
+
+    @Test
+    void aRepeatQuarantineCallRestartsTheCooldownAtFullLength() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.MINUTES.toNanos(30));
+        q.quarantine("shared-openai");
+
+        now.set(TimeUnit.MINUTES.toNanos(20));
+        q.quarantine("shared-openai");
+
+        now.set(TimeUnit.MINUTES.toNanos(45)); // 25 min after the second call, 45 after the first
+        assertTrue(q.isQuarantined("shared-openai"),
+                "a fresh exhaustion resets the cooldown to full length, not the earlier shorter wait");
+    }
+
+    @Test
+    void activeRemainingSecondsListsOnlyStillQuarantinedCredentials() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.MINUTES.toNanos(30));
+        q.quarantine("shared-openai");
+        q.quarantine("another-credential");
+
+        now.set(TimeUnit.MINUTES.toNanos(31));
+        q.quarantine("shared-openai"); // re-quarantined after the first one expired
+
+        Map<String, Long> active = q.activeRemainingSeconds();
+        assertEquals(Map.of("shared-openai", 1800L), active,
+                "the expired credential is dropped; the re-quarantined one is reported");
+    }
+
+    @Test
+    void noneReportsNothingQuarantinedWhenNeverToldTo() {
+        // .none() is the stand-in for a caller whose code path never calls #quarantine at all (e.g.
+        // the 2/5/6-arg CompositePeerLauncher constructors) — not a guarantee that a call to
+        // #quarantine on it is a no-op. Left alone, as those call sites leave it, nothing is ever
+        // quarantined.
+        BackendQuarantine q = BackendQuarantine.none();
+
+        assertFalse(q.isQuarantined("anything"));
+        assertTrue(q.activeRemainingSeconds().isEmpty());
+    }
+
+    @Test
+    void aNonPositiveCooldownIsRejected() {
+        assertThrows(IllegalArgumentException.class, () -> new BackendQuarantine(() -> 0L, 0L));
+        assertThrows(IllegalArgumentException.class, () -> new BackendQuarantine(() -> 0L, -1L));
+    }
+}

bridged/src/test/java/dev/ltms/bridged/placement/PlacementPolicyTest.java

@@ -24,7 +24,13 @@ class PlacementPolicyTest {
     private static PlacementContext ctx(List<PlacementCandidate> candidates,
                                         Function<String, Integer> liveCount,
                                         Set<String> unreachable) {
-        return new PlacementContext("b", candidates, liveCount, unreachable);
+        return ctx(candidates, liveCount, unreachable, Set.of());
+    }
+
+    private static PlacementContext ctx(List<PlacementCandidate> candidates,
+                                        Function<String, Integer> liveCount,
+                                        Set<String> unreachable, Set<String> quarantined) {
+        return new PlacementContext("b", candidates, liveCount, unreachable, quarantined);
     }
 
     private static PlacementContext ctx(List<PlacementCandidate> candidates,
@@ -46,17 +52,37 @@ class PlacementPolicyTest {
         PlacementPolicy policy = PlacementPolicies.fixed();
         PlacementContext ctx = new PlacementContext(null,
                 List.of(PlacementCandidate.profile("a"), PlacementCandidate.profile("b")),
-                noSessions(), Set.of());
+                noSessions(), Set.of(), Set.of());
         assertEquals("a", policy.select(ctx).profile());
     }
 
     @Test
     void fixedThrowsWhenNoProfilesAndNoDefault() {
         PlacementPolicy policy = PlacementPolicies.fixed();
-        PlacementContext ctx = new PlacementContext(null, List.of(), noSessions(), Set.of());
+        PlacementContext ctx = new PlacementContext(null, List.of(), noSessions(), Set.of(), Set.of());
         assertThrows(PlacementException.class, () -> policy.select(ctx));
     }
 
+    @Test
+    void fixedSkipsQuarantinedDefault() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a"), PlacementCandidate.profile("b")),
+                noSessions(), Set.of(), Set.of("b"));
+        assertEquals("a", policy.select(ctx).profile(),
+                "the default 'b' is quarantined, so fixed falls through to the first un-quarantined candidate");
+    }
+
+    @Test
+    void fixedThrowsWhenDefaultAndEveryCandidateQuarantined() {
+        PlacementPolicy policy = PlacementPolicies.fixed();
+        PlacementContext ctx = new PlacementContext("b",
+                List.of(PlacementCandidate.profile("a"), PlacementCandidate.profile("b")),
+                noSessions(), Set.of(), Set.of("a", "b"));
+        PlacementException e = assertThrows(PlacementException.class, () -> policy.select(ctx));
+        assertTrue(e.getMessage().contains("quarantined"), e.getMessage());
+    }
+
     @Test
     void roundRobinCyclesThroughAvailableProfiles() {
         PlacementPolicy policy = PlacementPolicies.roundRobin();
@@ -82,6 +108,18 @@ class PlacementPolicyTest {
         }
     }
 
+    @Test
+    void roundRobinSkipsQuarantinedProfiles() {
+        PlacementPolicy policy = PlacementPolicies.roundRobin();
+        List<PlacementCandidate> candidates = List.of(
+                PlacementCandidate.profile("a"),
+                PlacementCandidate.profile("b"));
+        PlacementContext ctx = ctx(candidates, noSessions(), Set.of(), Set.of("a"));
+        for (int i = 0; i < 5; i++) {
+            assertEquals("b", policy.select(ctx).profile(), "a is quarantined, so every pick lands on b");
+        }
+    }
+
     @Test
     void roundRobinThrowsWhenAllAtMaxLoad() {
         PlacementPolicy policy = PlacementPolicies.roundRobin();
@@ -150,6 +188,29 @@ class PlacementPolicyTest {
         assertTrue(e.getMessage().contains("maxLoad"), e.getMessage());
     }
 
+    @Test
+    void weightedSkipsQuarantinedProfile() {
+        PlacementPolicy policy = PlacementPolicies.weighted();
+        List<PlacementCandidate> candidates = List.of(
+                PlacementCandidate.profile("a", 1.0f, null),
+                PlacementCandidate.profile("b", 1.0f, null));
+        PlacementContext ctx = ctx(candidates, noSessions(), Set.of(), Set.of("a"));
+        for (int i = 0; i < 5; i++) {
+            assertEquals("b", policy.select(ctx).profile(), "a is quarantined, so every pick lands on b");
+        }
+    }
+
+    @Test
+    void weightedThrowsWhenAllQuarantined() {
+        PlacementPolicy policy = PlacementPolicies.weighted();
+        List<PlacementCandidate> candidates = List.of(
+                PlacementCandidate.profile("a"),
+                PlacementCandidate.profile("b"));
+        PlacementException e = assertThrows(PlacementException.class,
+                () -> policy.select(ctx(candidates, noSessions(), Set.of(), Set.of("a", "b"))));
+        assertTrue(e.getMessage().contains("quarantined"), e.getMessage());
+    }
+
     @Test
     void weightedThrowsWhenAllUnreachable() {
         PlacementPolicy policy = PlacementPolicies.weighted();