From e501d399889d2ef7c1f34ccedd058808d05a1a7a Mon Sep 17 00:00:00 2001 From: Dai Ha Date: Sat, 15 Aug 2026 10:29:54 +0200 Subject: [PATCH 1/2] CB-578 stage B: quarantine the exhausted credential, not the profile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/bridged.example.yaml | 40 +++++-- .../main/java/dev/ltms/bridged/Bridged.java | 36 +++++- .../ltms/bridged/config/BridgedConfig.java | 85 ++++++++++++-- .../dev/ltms/bridged/config/ConfigRef.java | 29 ++++- .../bridged/inject/CompletionResolver.java | 12 +- .../ltms/bridged/inject/ExhaustionSink.java | 31 +++++ .../java/dev/ltms/bridged/mcp/BridgeMcp.java | 57 +++++++-- .../bridged/member/CompositePeerLauncher.java | 93 +++++++++++++-- .../bridged/placement/BackendQuarantine.java | 99 ++++++++++++++++ .../placement/FixedPlacementPolicy.java | 18 ++- .../bridged/placement/PlacementContext.java | 6 +- .../placement/PlacementPolicyUtil.java | 19 ++- .../ltms/bridged/config/ConfigRefTest.java | 48 ++++++++ .../inject/CompletionResolverTest.java | 79 ++++++++++--- .../ltms/bridged/mcp/BridgeMcpAuthzTest.java | 3 +- .../dev/ltms/bridged/mcp/BridgeMcpTest.java | 21 +++- .../member/CompositePeerLauncherTest.java | 111 ++++++++++++++++++ .../ltms/bridged/msg/MessageServiceTest.java | 3 +- .../placement/BackendQuarantineTest.java | 108 +++++++++++++++++ .../placement/PlacementPolicyTest.java | 67 ++++++++++- 20 files changed, 886 insertions(+), 79 deletions(-) create mode 100644 bridged/src/main/java/dev/ltms/bridged/inject/ExhaustionSink.java create mode 100644 bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java create mode 100644 bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java diff --git a/bridged/bridged.example.yaml b/bridged/bridged.example.yaml index 3fcf852..e297fe9 100644 --- a/bridged/bridged.example.yaml +++ b/bridged/bridged.example.yaml @@ -144,6 +144,17 @@ herdrSocket: ~/.config/herdr/herdr.sock # omit and this profile's completion fallback behaves exactly as before. # Every backend words its refusal differently, so this is config, never a # vendor string baked into bridged itself. +# DEFERRED: compiled once into a startup pattern map — editing it needs a +# daemon restart, same as this profile's model/baseUrl/argv. +# credentialId → CB-578 stage B: the credential this profile quarantines WITH when a +# BACKEND_EXHAUSTED classification fires. Two profiles that set the SAME +# credentialId share one quarantine — the case this exists for is two models +# on one account (e.g. sol and terra both billing one OpenAI credential): an +# exhaustion on either one must lock out both, or the fleet just walks onto +# the same dead account under the sibling's name. Opt-in — omit and this +# profile quarantines alone, under its own name, exactly as if the field did +# not exist. Cooldown length is the top-level quarantineCooldownSeconds below. +# HOT: read live at every spawn/exhaustion check — no restart needed. # env → extra environment for this profile's workers, as a literal key/value map # (CB-511). Use it to give workers a toolchain. # @@ -178,6 +189,7 @@ profiles: # gitTokenEnv: GITEA_TOKEN # opt-in: let this profile's workers open their own PR (CB-302) # gitHostEnv: GITEA_HOST # defaults to GITEA_HOST; injected only with gitTokenEnv # exhaustedPattern: "usage limit has been reached" # opt-in: classify a usage-limit refusal (CB-578) + # credentialId: shared-openai # opt-in: quarantine together with every other profile sharing this id (CB-578) # configDir: /Users/me/.ccs/instances/gx10 # CLAUDE_CONFIG_DIR — inherit that profile's skills/MCP # cwd: /Users/me/src/myrepo # pin the working dir; omit to inherit the primary's # parityOverlay: [".claude/settings.local.json", ".env", ".envrc"] # never add .mcp.json — see above @@ -243,6 +255,15 @@ profiles: # round-robin, or weighted. Omitting this key is a strict no-op for existing configs. placement: weighted +# How long a credential sits out after a BACKEND_EXHAUSTED classification (CB-578 stage B), in +# seconds, before a spawn may land on it again. Applies to every profile's effective credential +# (its own name, or its credentialId if set above) — there is no per-profile override. Default +# 1800 (30 minutes) when omitted or non-positive. +# DEFERRED: baked once into the BackendQuarantine built at startup — a running quarantine keeps +# its original cooldown regardless; a new value only applies to a quarantine that starts after a +# restart. Editing this needs a daemon restart to take effect. +# quarantineCooldownSeconds: 1800 + # Re-read this file without restarting the daemon (CB-559). Off unless you add this block, so an # upgraded bridged keeps the old behaviour: the file is read once at boot and never again. # enabled → turn the watch on. bridged checks the file's modified time on a timer and @@ -252,17 +273,20 @@ placement: weighted # Not every key can move under a running daemon, and the difference is about what already exists # when the reload happens — not about how important the key is: # HOT → takes effect on the next spawn: the whole `fleet:` block (every role pool, -# `charters`, and `tabLabel`), `placement:`, and an existing profile's weight / maxLoad. Those are -# hot because the placement policy reads them through a supplier — being config is +# `charters`, and `tabLabel`), `placement:`, and an existing profile's weight / maxLoad +# / credentialId. Those are hot because the placement policy (and, for credentialId, +# the CB-578 stage B quarantine check) reads them through a supplier — being config is # not by itself enough to make a key hot. # DEFERRED → accepted into the new config, but the wiring built at startup keeps the old value # until you restart: `lifecycle:`, `leadHeartbeat:`, `guard:`, `worktreeRoot:`, -# `spawnReadyTimeoutMs` / `spawnReadyPollMs`, ADDING or REMOVING a profile (a new -# backend needs its own launcher, and launchers are built once), AND an existing -# profile's launch settings — model, baseUrl, argv, env, configDir, mcpUrl, tabLabel. -# The launcher takes a copy of `profiles:` at startup and resolves every spawn out of -# that copy, so those never reach a launch until you restart. The reload logs them by -# name rather than pretending they applied. +# `spawnReadyTimeoutMs` / `spawnReadyPollMs`, `quarantineCooldownSeconds` (CB-578 +# stage B — baked once into the quarantine tracker built at startup), ADDING or +# REMOVING a profile (a new backend needs its own launcher, and launchers are built +# once), AND an existing profile's launch settings — model, baseUrl, argv, env, +# configDir, mcpUrl, tabLabel, exhaustedPattern. The launcher takes a copy of +# `profiles:` at startup and resolves every spawn out of that copy, so those never +# reach a launch until you restart. The reload logs them by name rather than +# pretending they applied. # COLD → cannot change at all: `bind:`, `herdrSocket:`, `broker:` and `auth:`. The socket is # bound, the broker connection is open, and the auth mode decides who may reach the # port that is already listening. diff --git a/bridged/src/main/java/dev/ltms/bridged/Bridged.java b/bridged/src/main/java/dev/ltms/bridged/Bridged.java index 97211d2..797ad3b 100644 --- a/bridged/src/main/java/dev/ltms/bridged/Bridged.java +++ b/bridged/src/main/java/dev/ltms/bridged/Bridged.java @@ -14,6 +14,7 @@ import dev.ltms.bridged.herdr.UnixSocketHerdrClient; import dev.ltms.bridged.herdr.WorkspaceControl; import dev.ltms.bridged.inject.CompletionResolver; import dev.ltms.bridged.inject.ExhaustedPatternLookup; +import dev.ltms.bridged.inject.ExhaustionSink; import dev.ltms.bridged.inject.Injector; import dev.ltms.bridged.inject.StatusPoller; import dev.ltms.bridged.inject.TurnListener; @@ -37,6 +38,7 @@ import dev.ltms.bridged.msg.LeadHeartbeatLoop; import dev.ltms.bridged.msg.ReplyPushLoop; import dev.ltms.bridged.rest.BridgedApp; import dev.ltms.bridged.session.GitWorktrees; +import dev.ltms.bridged.session.MemberSession; import dev.ltms.bridged.session.SessionManager; import dev.ltms.bridged.peer.PeerLauncher; import dev.ltms.bridged.session.SessionReaper; @@ -44,6 +46,7 @@ import dev.ltms.bridged.member.ClaudeCodeLauncher; import dev.ltms.bridged.member.CompositePeerLauncher; import dev.ltms.bridged.member.HerdrPeerLauncher; import dev.ltms.bridged.member.OpenCodeLauncher; +import dev.ltms.bridged.placement.BackendQuarantine; import io.javalin.Javalin; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -141,11 +144,18 @@ public final class Bridged { () -> config.get().fleet())); } AtomicReference> liveCountRef = new AtomicReference<>(_ -> 0); + // CB-578 stage B: one quarantine tracker for the whole daemon, shared between the launcher + // (checked at spawn) and the exhaustion sink wired in below (written on BACKEND_EXHAUSTED). + // The cooldown is deferred (see BridgedConfig#quarantineCooldownSeconds): it is read once + // here, at startup, and a config reload only changes it for a daemon restart. + BackendQuarantine quarantine = new BackendQuarantine(System::nanoTime, + TimeUnit.SECONDS.toNanos(cfg.quarantineCooldownSeconds())); PeerLauncher workers = new CompositePeerLauncher( adapters, cfg.effectiveDefaultProfile(), config, - profileName -> liveCountRef.get().apply(profileName)); + profileName -> liveCountRef.get().apply(profileName), + quarantine); // CB-504: under supervision (launchd/systemd) bridged can start before herdr's socket // exists. The client itself is lazy — it connects per call — but the orphan reap below is // the first thing that actually talks to herdr, so without this wait a boot-order race @@ -272,7 +282,23 @@ public final class Bridged { .orElse(null); log.info("backend-exhausted classification (CB-578 stage A): {}", CompletionResolver.coverage(cfg.profiles().keySet(), exhaustedPatternsByProfile.keySet())); - CompletionResolver completion = new CompletionResolver(agents, rendezvous, exhaustedPatterns); + // CB-578 stage B: on a classification that actually wins, quarantine the exhausted profile's + // CREDENTIAL — not the profile name — so a profile sharing that credential (e.g. two models + // on one OpenAI account) is refused too, not just the one that happened to report it. Reads + // the profile config live off `config`, so a credentialId edit is hot: no restart needed. + ExhaustionSink exhaustionSink = (target, reason) -> sessions.roster().stream() + .filter(session -> target.equals(session.terminalId())) + .findFirst() + .map(MemberSession::profile) + .map(profileName -> config.get().profiles().get(profileName)) + .ifPresent(profile -> { + String credentialId = profile.effectiveCredentialId(); + quarantine.quarantine(credentialId); + log.warn("credential '{}' quarantined for {}s (profile '{}' classified " + + "BACKEND_EXHAUSTED): {}", credentialId, + cfg.quarantineCooldownSeconds(), profile.profile(), reason); + }); + CompletionResolver completion = new CompletionResolver(agents, rendezvous, exhaustedPatterns, exhaustionSink); // CB-113: deliver only to an available worker (its MCP is connected), never its boot window. // CB-301: the manager's presence bridge records availability and drives SPAWNING → READY. MemberPresence presence = sessions.asPresence(); @@ -440,7 +466,11 @@ public final class Bridged { var health = config.get().health(); return FleetHealthMonitor.coverage(health != null && health.isEnabled(), health != null && health.notifications() != null && health.notifications().configured()); - })); + }), + new BridgeMcp.QuarantineSource(profile -> { + var configured = config.get().profiles().get(profile); + return configured == null ? null : configured.effectiveCredentialId(); + }, quarantine)); // CB-559: opt-in config reload. With no `configReload:` block nothing is constructed, so an // upgraded daemon behaves exactly as before — the file is read once at boot and never again. diff --git a/bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java b/bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java index 4d257a9..6ff7fc5 100644 --- a/bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java +++ b/bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java @@ -58,6 +58,12 @@ import java.util.Set; * {@code fixed} (default), {@code round-robin}, or {@code weighted} * @param auth API authentication mode ({@code null} → {@code loopback-trust}, the * historical behaviour), CB-501 + * @param quarantineCooldownSeconds how long a credential stays quarantined after a + * {@code BACKEND_EXHAUSTED} classification (CB-578 stage B); {@code null}/{@code + * <=0} → {@link #DEFAULT_QUARANTINE_COOLDOWN_SECONDS}. Baked once into the + * {@code BackendQuarantine} built at startup, so it is DEFERRED: changing it + * needs a restart, and a quarantine already running keeps whatever cooldown was + * live when it started. */ @JsonIgnoreProperties(ignoreUnknown = true) public record BridgedConfig( @@ -76,7 +82,11 @@ public record BridgedConfig( Health health, String placement, Auth auth, - ConfigReload configReload) { + ConfigReload configReload, + Integer quarantineCooldownSeconds) { + + /** Default cooldown (CB-578 stage B) when {@code quarantineCooldownSeconds} is absent/non-positive. */ + public static final int DEFAULT_QUARANTINE_COOLDOWN_SECONDS = 1800; /** Back-compat 14-arg form — no {@code configReload:} block, so file watching stays off. */ public BridgedConfig(Bind bind, String herdrSocket, Map profiles, Guard guard, @@ -84,7 +94,7 @@ public record BridgedConfig( Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet, LeadHeartbeat leadHeartbeat, String placement, Auth auth) { this(bind, herdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs, - spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, null); + spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, null, null); } /** Back-compat form before the optional {@code health:} block was added. */ @@ -93,7 +103,18 @@ public record BridgedConfig( Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet, LeadHeartbeat leadHeartbeat, String placement, Auth auth, ConfigReload configReload) { this(bind, herdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs, - spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, configReload); + spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, null, placement, auth, configReload, null); + } + + /** Back-compat form before the CB-578 stage B {@code quarantineCooldownSeconds} field was added. */ + public BridgedConfig(Bind bind, String herdrSocket, Map profiles, Guard guard, + String worktreeRoot, Lifecycle lifecycle, Integer spawnReadyTimeoutMs, + Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet, + LeadHeartbeat leadHeartbeat, Health health, String placement, Auth auth, + ConfigReload configReload) { + this(bind, herdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs, + spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, health, placement, auth, + configReload, null); } /** @@ -187,6 +208,17 @@ public record BridgedConfig( * {@code null}/blank ⇒ the classification never fires for this profile and * today's completion-fallback behaviour is unchanged. Every backend words * its refusal differently, so this is config, never a vendor string in code. + * @param credentialId the shared account this profile authenticates as (CB-578 stage B). Two + * or more profiles setting the same non-blank value are quarantined + * together by one {@code exhaustedPattern} classification on any one of + * them — the case a single OpenAI (or any other) credential backing several + * profiles ({@code sol}, {@code terra}, ...) needs, so a spawn does not walk + * straight onto the other profile sharing the same exhausted account. + * {@code null}/blank ⇒ this profile's {@link #effectiveCredentialId()} is + * its own name, so it quarantines alone — today's behaviour for every + * profile that does not opt in. Read live off the current config, so it is + * HOT: a change takes effect on the next exhaustion classification / spawn, + * no restart needed. */ @JsonIgnoreProperties(ignoreUnknown = true) public record Profile(String profile, String baseUrl, String model, @@ -200,7 +232,8 @@ public record BridgedConfig( Float weight, Integer maxLoad, Boolean subscription, - String exhaustedPattern) { + String exhaustedPattern, + String credentialId) { /** Peer kind spawned by {@link dev.ltms.bridged.member.ClaudeCodeLauncher} (the default). */ public static final String KIND_CLAUDE_CODE = "claude-code"; @@ -246,6 +279,10 @@ public record BridgedConfig( // exhaustedPattern stays null when unset/blank (opt-in) — no defaulting, no vendor // wording: an unconfigured profile keeps today's completion-fallback behaviour exactly. exhaustedPattern = (exhaustedPattern == null || exhaustedPattern.isBlank()) ? null : exhaustedPattern; + // credentialId stays null when unset/blank — effectiveCredentialId() is where the + // "quarantines alone" fallback actually lives, so today's behaviour needs no defaulting + // here at all. + credentialId = (credentialId == null || credentialId.isBlank()) ? null : credentialId; } /** @@ -290,7 +327,7 @@ public record BridgedConfig( public Profile withProfile(String p) { return new Profile(p, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel, mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, subscription, - exhaustedPattern); + exhaustedPattern, credentialId); } /** True when this profile is served by the Claude Code adapter (the default kind). */ @@ -326,11 +363,38 @@ public record BridgedConfig( mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, null, null); } + /** + * Backward-compatible constructor without the CB-578 stage B {@code credentialId} — the + * profile quarantines alone under its own name (see {@link #effectiveCredentialId()}). Keeps + * pre-stage-B call sites (and any YAML that omits the key) compiling and behaving identically. + */ + public Profile(String profile, String baseUrl, String model, + String configDir, String tokenEnv, List argv, + String placement, String workspace, String tabLabel, String mcpUrl, + String cwd, List parityOverlay, String gitTokenEnv, String gitHostEnv, + String kind, Map env, Float weight, Integer maxLoad, + Boolean subscription, String exhaustedPattern) { + this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel, + mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, + subscription, exhaustedPattern, null); + } + /** True when this profile's CB-578 stage A backend-exhausted classification is configured. */ public boolean hasExhaustedPattern() { return exhaustedPattern != null; } + /** + * The credential group this profile quarantines with (CB-578 stage B): the configured + * {@link #credentialId} when set, else this profile's own name — so an unconfigured profile + * quarantines alone, exactly as it did before this field existed. Two profiles that set the + * same non-blank {@code credentialId} share one quarantine: a {@code BACKEND_EXHAUSTED} + * classification on either one quarantines both. + */ + public String effectiveCredentialId() { + return (credentialId == null || credentialId.isBlank()) ? profile : credentialId; + } + /** True when this profile's workers are granted a forge token to open their own PR (CB-302). */ public boolean hasGitToken() { return gitTokenEnv != null && !gitTokenEnv.isBlank(); @@ -850,7 +914,7 @@ public record BridgedConfig( private static final Set KNOWN_TOP_LEVEL_KEYS = Set.of( "bind", "herdrSocket", "profiles", "guard", "worktreeRoot", "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs", "broker", "primary", "fleet", - "leadHeartbeat", "health", "placement", "auth", "configReload"); + "leadHeartbeat", "health", "placement", "auth", "configReload", "quarantineCooldownSeconds"); /** Load and validate config from {@code path}. */ public static BridgedConfig load(Path path) { @@ -1157,8 +1221,15 @@ public record BridgedConfig( // configReload is left as-is: null is "off", and ConfigReload's own compact constructor // defaults the fields of a block that IS present. Defaulting it here would start watching // the file for every config that never asked to be watched. + // quarantineCooldownSeconds IS defaulted, unlike leadHeartbeat/configReload above: it has no + // separate on/off switch of its own (BackendQuarantine only ever quarantines a credential + // after a BACKEND_EXHAUSTED classification, which stays opt-in via exhaustedPattern), so a + // config that never mentions it should still get a sane cooldown rather than a null one. + Integer quarantineCooldown = (quarantineCooldownSeconds != null && quarantineCooldownSeconds > 0) + ? quarantineCooldownSeconds : DEFAULT_QUARANTINE_COOLDOWN_SECONDS; return new BridgedConfig(b, herdrSocket, profiles, g, worktreeRoot, l, timeout, pollMs, - broker, primary, f, leadHeartbeat, health, placementOrDefault, a, configReload); + broker, primary, f, leadHeartbeat, health, placementOrDefault, a, configReload, + quarantineCooldown); } /** diff --git a/bridged/src/main/java/dev/ltms/bridged/config/ConfigRef.java b/bridged/src/main/java/dev/ltms/bridged/config/ConfigRef.java index d145555..f802e42 100644 --- a/bridged/src/main/java/dev/ltms/bridged/config/ConfigRef.java +++ b/bridged/src/main/java/dev/ltms/bridged/config/ConfigRef.java @@ -37,10 +37,15 @@ import java.util.function.Supplier; * {@code fleet.leaders} needs a restart, the same as any deferred key below. *
  • Deferred — accepted into the new snapshot, but the wiring built at startup * keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:}, - * {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code guard:}, - * {@code worktreeRoot:}, adding or removing a profile (a new backend needs its own launcher, + * {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code quarantineCooldownSeconds} + * (CB-578 stage B — baked once into the {@code BackendQuarantine} built at startup), + * {@code guard:}, {@code worktreeRoot:}, adding or removing a profile (a new backend needs its own launcher, * which is constructed once), and an existing profile's launch settings — - * {@code model}, {@code baseUrl}, {@code argv}, {@code env}, {@code mcpUrl} and the rest. + * {@code model}, {@code baseUrl}, {@code argv}, {@code env}, {@code mcpUrl}, + * {@code exhaustedPattern} (CB-578 stage A — compiled once into {@code Bridged.main}'s + * pattern map at startup), and the rest. {@code credentialId} (CB-578 stage B) is NOT on + * this list — it is read live off the config supplier at every quarantine check and + * exhaustion event, exactly like {@code weight} / {@code maxLoad}, so it is hot instead. * {@code HerdrPeerLauncher} takes {@code Map.copyOf(profiles)} at construction and resolves * each spawn out of that copy, so those never reach a launch until the daemon restarts. A * reload logs these rather than pretending they applied.
    • @@ -215,6 +220,12 @@ public final class ConfigRef implements Supplier { || !Objects.equals(old.spawnReadyPollMs(), fresh.spawnReadyPollMs())) { changed.add("spawnReady*"); } + // CB-578 stage B: baked once into the BackendQuarantine built at startup — a running + // quarantine keeps its original cooldown regardless, and a new cooldown only applies to a + // quarantine that starts after a restart. + if (!Objects.equals(old.quarantineCooldownSeconds(), fresh.quarantineCooldownSeconds())) { + changed.add("quarantineCooldownSeconds"); + } Map before = old.profiles() == null ? Map.of() : old.profiles(); Map after = @@ -250,8 +261,10 @@ public final class ConfigRef implements Supplier { /** * Whether two versions of a profile would launch a peer identically. Compares every component - * the launcher reads at spawn; {@code weight} and {@code maxLoad} are excluded because those are - * read live by the placement policy and really do take effect on the next spawn. + * the launcher reads at spawn; {@code weight}, {@code maxLoad} and {@code credentialId} are + * excluded because those are read live (by the placement policy and, for credentialId, by + * {@code CompositePeerLauncher}/the CB-578 stage B exhaustion sink) and really do take effect on + * the next spawn. */ private static boolean sameLaunchSettings(BridgedConfig.Profile a, BridgedConfig.Profile b) { return Objects.equals(a.baseUrl(), b.baseUrl()) @@ -269,6 +282,10 @@ public final class ConfigRef implements Supplier { && Objects.equals(a.gitHostEnv(), b.gitHostEnv()) && Objects.equals(a.kind(), b.kind()) && Objects.equals(a.env(), b.env()) - && Objects.equals(a.subscription(), b.subscription()); + && Objects.equals(a.subscription(), b.subscription()) + // CB-578 stage B: exhaustedPattern is compiled once into Bridged.main's pattern map + // at startup (see ExhaustedPatternLookup wiring) — a reload never re-reads it, so a + // changed pattern must be reported as deferred, exactly like model/baseUrl/argv. + && Objects.equals(a.exhaustedPattern(), b.exhaustedPattern()); } } diff --git a/bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java b/bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java index 82c0b6a..9a44702 100644 --- a/bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java +++ b/bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java @@ -67,6 +67,7 @@ public final class CompletionResolver implements TurnListener { private final AgentControl agents; private final Rendezvous rendezvous; private final ExhaustedPatternLookup exhaustedPatterns; + private final ExhaustionSink exhaustionSink; /** * Per-target record of the turn currently in flight: the exact {@link Rendezvous} waiter its @@ -89,11 +90,17 @@ public final class CompletionResolver implements TurnListener { * usage-limit refusal pattern. Required — there is deliberately no * defaulting overload; a caller that does not want the classification * must pass an explicit inert value ({@link ExhaustedPatternLookup#none()}). + * @param exhaustionSink CB-578 stage B: notified when a {@code BACKEND_EXHAUSTED} + * classification actually resolves a waiter. Required for the same + * reason as {@code exhaustedPatterns} — pass {@link ExhaustionSink#none()} + * to opt out. */ - public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns) { + public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns, + ExhaustionSink exhaustionSink) { this.agents = agents; this.rendezvous = rendezvous; this.exhaustedPatterns = Objects.requireNonNull(exhaustedPatterns, "exhaustedPatterns"); + this.exhaustionSink = Objects.requireNonNull(exhaustionSink, "exhaustionSink"); } @Override @@ -210,6 +217,9 @@ public final class CompletionResolver implements TurnListener { inFlight.remove(target, turn); log.warn("completion for {} classified BACKEND_EXHAUSTED (no bridge_reply; scrape " + "matched the profile's exhausted pattern): {}", target, reason); + // CB-578 stage B: only on the resolution that actually won the race — a late + // duplicate must never quarantine a credential twice for one refusal. + exhaustionSink.onExhausted(target, reason); } return; } diff --git a/bridged/src/main/java/dev/ltms/bridged/inject/ExhaustionSink.java b/bridged/src/main/java/dev/ltms/bridged/inject/ExhaustionSink.java new file mode 100644 index 0000000..d7b35a0 --- /dev/null +++ b/bridged/src/main/java/dev/ltms/bridged/inject/ExhaustionSink.java @@ -0,0 +1,31 @@ +package dev.ltms.bridged.inject; + +/** + * Notified when {@link CompletionResolver} actually delivers a {@code BACKEND_EXHAUSTED} + * classification to a waiting send (CB-578 stage B) — never on a race that lost (see + * {@link CompletionResolver#resolve}, which only calls this after + * {@code Rendezvous.resolveExhausted} returns {@code true}). + * + *

    {@link CompletionResolver} knows only {@code target} (a herdr terminal id); it has no notion of + * profiles or credentials, so mapping {@code target} to whatever should be quarantined is entirely + * the sink's job — see {@code Bridged.main}'s wiring, which resolves target → session → profile → + * {@code effectiveCredentialId()} and calls {@code BackendQuarantine.quarantine} on it. + */ +@FunctionalInterface +public interface ExhaustionSink { + + /** + * @param target the herdr terminal id whose turn was classified {@code BACKEND_EXHAUSTED} + * @param reason the matched-line reason carried by the classification + */ + void onExhausted(String target, String reason); + + /** + * Inert sink — nothing happens on exhaustion. The explicit stand-in a caller (or a test not + * exercising this feature) passes instead of a defaulting overload, exactly like + * {@link ExhaustedPatternLookup#none()}. + */ + static ExhaustionSink none() { + return (target, reason) -> { }; + } +} diff --git a/bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java b/bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java index da37ea9..0527bc7 100644 --- a/bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java +++ b/bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java @@ -14,6 +14,7 @@ import dev.ltms.bridged.herdr.HerdrException; import dev.ltms.bridged.msg.MessageService; import dev.ltms.bridged.msg.Rendezvous; import dev.ltms.bridged.peer.PeerUnreachableException; +import dev.ltms.bridged.placement.BackendQuarantine; import dev.ltms.bridged.session.SessionManager; import dev.ltms.bridged.session.MemberSession; import dev.ltms.bridged.session.WorktreeRequest; @@ -33,6 +34,7 @@ import jakarta.servlet.http.HttpServlet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.function.Function; import java.util.function.LongSupplier; @@ -78,6 +80,7 @@ public final class BridgeMcp { private final Metrics metrics; // CB-502: null → auth failures not counted private final CapacitySource capacity; private final HealthCoverageSource healthCoverage; + private final QuarantineSource quarantine; /** Capacity facts used by {@code bridge_list}; production must supply the placement live count. */ public record CapacitySource(Function liveCount, Function maxLoad, @@ -90,17 +93,30 @@ public final class BridgeMcp { /** Coverage is supplied by the health wiring, not inferred from a missing dependency. */ public record HealthCoverageSource(Supplier value) { } + /** + * CB-578 stage B quarantine facts used by {@code bridge_profiles}: a profile → credential id + * lookup, plus the shared {@link BackendQuarantine} to read remaining cooldowns off. + */ + public record QuarantineSource(Function credentialIdFor, BackendQuarantine quarantine) { + /** Inert source — no profile is ever reported quarantined. Explicit stand-in, not a default. */ + public static QuarantineSource none() { return new QuarantineSource(_ -> null, BackendQuarantine.none()); } + } + /** * @param callers resolves each call's {@link Principal}; {@code null} disables authorization. * This surface needs its own enforcement: {@code /mcp} is a raw servlet on * Jetty's context handler and never passes through Javalin's {@code before} * filter, so the REST guard does not cover it. * @param metrics registry for auth-failure counting; may be {@code null} + * @param quarantine CB-578 stage B facts for {@code bridge_profiles}; required — pass + * {@link QuarantineSource#none()} for a caller that does not want the feature */ public BridgeMcp(MessageService messages, PeerLauncher workers, SessionManager sessions, ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry, - CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage) { + CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage, + QuarantineSource quarantine) { this.capacity = capacity; + this.quarantine = Objects.requireNonNull(quarantine, "quarantine"); this.healthCoverage = healthCoverage; McpJsonMapper json = new JacksonMcpJsonMapperSupplier().get(); this.transport = HttpServletStreamableServerTransportProvider.builder() @@ -228,7 +244,7 @@ public final class BridgeMcp { .toolCall(profilesTool(), (exchange, _) -> { McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null); if (denied != null) return denied; - return profiles(workers); + return profiles(workers, quarantine); }) .toolCall(whoamiTool(), (exchange, _) -> { McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null); @@ -706,11 +722,33 @@ public final class BridgeMcp { return null; } - /** {@code bridge_profiles}: the configured worker profiles and the default. */ - static McpSchema.CallToolResult profiles(PeerLauncher workers) { - return text(json(Map.of( - "profiles", workers.profiles(), - "default", workers.defaultProfile() == null ? "" : workers.defaultProfile()))); + /** + * {@code bridge_profiles}: the configured worker profiles, the default, and — CB-578 stage B — + * which of them are currently quarantined (backend exhausted) and for how much longer. The + * {@code quarantined} key is present only when at least one profile is, so a fleet where nothing + * has ever been quarantined gets exactly the pre-stage-B shape. + */ + static McpSchema.CallToolResult profiles(PeerLauncher workers, QuarantineSource quarantine) { + Map result = new LinkedHashMap<>(); + result.put("profiles", workers.profiles()); + result.put("default", workers.defaultProfile() == null ? "" : workers.defaultProfile()); + Map quarantined = new LinkedHashMap<>(); + for (String profile : workers.profiles()) { + String credentialId = quarantine.credentialIdFor().apply(profile); + if (credentialId == null) { + continue; + } + quarantine.quarantine().remainingSeconds(credentialId).ifPresent(remaining -> { + Map row = new LinkedHashMap<>(); + row.put("credentialId", credentialId); + row.put("quarantinedForSeconds", remaining); + quarantined.put(profile, row); + }); + } + if (!quarantined.isEmpty()) { + result.put("quarantined", quarantined); + } + return text(json(result)); } /** @@ -940,7 +978,10 @@ public final class BridgeMcp { private static McpSchema.Tool profilesTool() { return tool("bridge_profiles", - "List the configured worker profiles (backends) and which one bridge_spawn uses by default.", + "List the configured worker profiles (backends) and which one bridge_spawn uses by " + + "default. A 'quarantined' map is present when a backend-exhausted refusal put " + + "a profile's credential on cooldown — bridge_spawn onto it is refused until " + + "quarantinedForSeconds elapses; a profile sharing that credential is listed too.", objectSchema(Map.of(), List.of())); } diff --git a/bridged/src/main/java/dev/ltms/bridged/member/CompositePeerLauncher.java b/bridged/src/main/java/dev/ltms/bridged/member/CompositePeerLauncher.java index 605e0c2..73e08b4 100644 --- a/bridged/src/main/java/dev/ltms/bridged/member/CompositePeerLauncher.java +++ b/bridged/src/main/java/dev/ltms/bridged/member/CompositePeerLauncher.java @@ -8,6 +8,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.PlacementCandidate; import dev.ltms.bridged.placement.PlacementContext; import dev.ltms.bridged.placement.PlacementException; @@ -23,10 +24,12 @@ import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; /** * The {@link PeerLauncher} the core actually holds when more than one adapter is configured — a thin @@ -78,6 +81,9 @@ public final class CompositePeerLauncher implements PeerLauncher { private final Supplier> profileConfigs; private final Supplier placementPolicy; + /** CB-578 stage B: credential cooldown, checked before an explicit spawn and filtered into placement. */ + private final BackendQuarantine quarantine; + /** * CB-557: the role pools an unqualified spawn draws its candidates from. A supplier that yields * {@code null}, and an empty pool for a role, both fall back to every configured profile — the @@ -99,7 +105,10 @@ public final class CompositePeerLauncher implements PeerLauncher { } /** - * Production constructor with a placement policy and live-worker counter. + * Production constructor with a placement policy and live-worker counter. Quarantine (CB-578 + * stage B) is off for this constructor — {@link BackendQuarantine#none()} — since it predates + * the feature and existing callers of this exact overload never exercised it; use the 7-arg + * overload below to wire a real {@link BackendQuarantine}. * * @param delegates one adapter per configured peer kind; must be non-empty and declare * disjoint profile-name sets @@ -113,13 +122,14 @@ public final class CompositePeerLauncher implements PeerLauncher { Map profileConfigs, PlacementPolicy placementPolicy, Function liveCount) { - this(delegates, defaultProfile, profileConfigs, placementPolicy, liveCount, null); + this(delegates, defaultProfile, profileConfigs, placementPolicy, liveCount, null, BackendQuarantine.none()); } /** * Production constructor with role pools (CB-557). An unqualified spawn draws its candidates from * {@code fleet.} instead of from every configured profile, so a reviewer is placed on a - * reviewer backend and never on, say, the architect-only one. + * reviewer backend and never on, say, the architect-only one. Quarantine is off for this + * constructor too, for the same reason as the 5-arg overload above. * * @param fleet the configured role pools; {@code null} ⇒ every profile is a candidate for every * role, which is the pre-CB-557 behaviour @@ -130,29 +140,50 @@ public final class CompositePeerLauncher implements PeerLauncher { PlacementPolicy placementPolicy, Function liveCount, BridgedConfig.Fleet fleet) { + this(delegates, defaultProfile, profileConfigs, placementPolicy, liveCount, fleet, BackendQuarantine.none()); + } + + /** + * Production constructor with role pools and quarantine (CB-578 stage B). The full-featured + * non-reloading form; {@link #CompositePeerLauncher(List, String, Supplier, Function, BackendQuarantine)} + * is what {@code Bridged.main} actually wires up. + * + * @param quarantine required — pass {@link BackendQuarantine#none()} for a caller that does not + * want the feature, never a defaulting overload (CB-578 stage B's own rule). + */ + public CompositePeerLauncher(List delegates, + String defaultProfile, + Map profileConfigs, + PlacementPolicy placementPolicy, + Function liveCount, + BridgedConfig.Fleet fleet, + BackendQuarantine quarantine) { // LinkedHashMap, not Map.copyOf: candidates() promises definition order and the weighted // policy breaks exact-weight ties on it, so a salted iteration order would make placement // differ from one JVM run to the next. this(delegates, defaultProfile, constant(Collections.unmodifiableMap(new LinkedHashMap<>(profileConfigs))), - constant(placementPolicy), liveCount, constant(fleet)); + constant(placementPolicy), liveCount, constant(fleet), quarantine); } /** * Production constructor that re-reads its placement inputs per spawn (CB-559), so a config * reload retargets the next member without a restart. * - * @param config the live configuration — read at every spawn, never captured + * @param config the live configuration — read at every spawn, never captured + * @param quarantine required — CB-578 stage B; pass {@link BackendQuarantine#none()} to opt out */ public CompositePeerLauncher(List delegates, String defaultProfile, Supplier config, - Function liveCount) { + Function liveCount, + BackendQuarantine quarantine) { this(delegates, defaultProfile, () -> config.get().profiles(), () -> PlacementPolicies.fromName(config.get().placement()), liveCount, - () -> config.get().fleet()); + () -> config.get().fleet(), + quarantine); } /** The all-suppliers form every other constructor funnels into. */ @@ -161,8 +192,10 @@ public final class CompositePeerLauncher implements PeerLauncher { Supplier> profileConfigs, Supplier placementPolicy, Function liveCount, - Supplier fleet) { + Supplier fleet, + BackendQuarantine quarantine) { this.fleet = fleet; + this.quarantine = Objects.requireNonNull(quarantine, "quarantine"); if (delegates.isEmpty()) { throw new IllegalArgumentException("at least one peer adapter must be configured"); } @@ -225,6 +258,7 @@ public final class CompositePeerLauncher implements PeerLauncher { // the charter makes explicit-profile spawns the normal path — so skipping the check // here would leave the cap dead config in real operation. HerdrPeerLauncher d = route(requestedProfile); + enforceNotQuarantined(requestedProfile); enforceMaxLoad(requestedProfile); PeerHandle handle = d.spawn(req); spawnedBy.put(handle.id(), d); @@ -238,7 +272,10 @@ public final class CompositePeerLauncher implements PeerLauncher { List candidates = candidates(req.role()); String roleDefault = defaultProfileFor(req.role()); Set unreachable = new HashSet<>(); - PlacementContext ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable); + // CB-578 stage B: computed once up front — a quarantine's expiry cannot pass within one spawn + // call, so re-deriving it per retry would only cost work, never change the answer. + Set quarantined = quarantinedProfiles(candidates); + PlacementContext ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable, quarantined); int maxAttempts = candidates.isEmpty() ? 1 : candidates.size(); for (int attempt = 0; attempt < maxAttempts; attempt++) { @@ -268,7 +305,7 @@ public final class CompositePeerLauncher implements PeerLauncher { chosen.profile(), e.getMessage()); unreachable.add(chosen.profile()); // Update the context for the next selection so the policy excludes this profile. - ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable); + ctx = new PlacementContext(roleDefault, candidates, liveCount, unreachable, quarantined); } } @@ -298,6 +335,42 @@ public final class CompositePeerLauncher implements PeerLauncher { * @param profile the profile the caller explicitly named * @throws PlacementException when the profile is at capacity */ + /** + * Refuse an explicit-profile spawn whose credential is quarantined (CB-578 stage B): a prior + * {@code BACKEND_EXHAUSTED} classification on this profile, or on another profile sharing its + * {@code credentialId}, is still on cooldown. + * + *

    Checked before {@link #enforceMaxLoad}, and for the same reason that check exists: an + * explicit profile bypasses the placement policy's filtering entirely, so without this the cap + * (there, quarantine here) would be dead config on the very path the charter calls normal. + * Deliberately no fallback to another profile, matching {@link #enforceMaxLoad}'s own reasoning — + * the caller named this profile for a cost/model reason. + * + * @throws PlacementException naming the profile, its credential, and the remaining cooldown + */ + private void enforceNotQuarantined(String profile) { + String credentialId = credentialIdFor(profile); + quarantine.remainingSeconds(credentialId).ifPresent(remaining -> { + throw new PlacementException("worker profile '" + profile + "' is quarantined " + + "(credential '" + credentialId + "' exhausted; ~" + remaining + + "s remaining) — refusing spawn"); + }); + } + + /** {@code profile}'s credential group (CB-578 stage B), or the profile's own name if unconfigured. */ + private String credentialIdFor(String profile) { + BridgedConfig.Profile cfg = profiles0().get(profile); + return cfg == null ? profile : cfg.effectiveCredentialId(); + } + + /** The subset of {@code candidates} whose credential is currently quarantined (CB-578 stage B). */ + private Set quarantinedProfiles(List candidates) { + return candidates.stream() + .map(PlacementCandidate::profile) + .filter(p -> quarantine.isQuarantined(credentialIdFor(p))) + .collect(Collectors.toSet()); + } + private void enforceMaxLoad(String profile) { // Absent config, or a config whose maxLoad normalized to null (non-positive ⇒ unlimited at // load), means no cap — never cap what wasn't configured. diff --git a/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java b/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java new file mode 100644 index 0000000..1917a58 --- /dev/null +++ b/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java @@ -0,0 +1,99 @@ +package dev.ltms.bridged.placement; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import java.util.OptionalLong; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.LongSupplier; + +/** + * Where a credential (not a profile — see {@code BridgedConfig.Profile#effectiveCredentialId()}) + * sits out a cooldown after a {@code BACKEND_EXHAUSTED} classification (CB-578 stage B), so a fresh + * spawn does not walk straight back onto the account that just refused on a usage limit. + * + *

    Keyed by credential id, never by profile name: two profiles sharing one credential (e.g. two + * models on the same OpenAI account) share one quarantine — {@link #quarantine} one credential id + * and every profile whose {@code effectiveCredentialId()} equals it is quarantined too, without this + * class knowing anything about profiles at all. That mapping is the caller's job (see + * {@code CompositePeerLauncher} and {@code dev.ltms.bridged.inject.ExhaustionSink}). + * + *

    The clock is injected ({@link LongSupplier}, conventionally {@code System::nanoTime} like + * {@code FleetHealthMonitor}), never read inline, so a quarantine's expiry is testable without a + * real sleep. + */ +public final class BackendQuarantine { + + private final ConcurrentHashMap quarantinedUntilNanos = new ConcurrentHashMap<>(); + private final LongSupplier nowNanos; + private final long cooldownNanos; + + /** + * @param nowNanos monotonic clock, injected for testability + * @param cooldownNanos how long a fresh {@link #quarantine} call blocks the credential for; + * must be positive + */ + public BackendQuarantine(LongSupplier nowNanos, long cooldownNanos) { + this.nowNanos = Objects.requireNonNull(nowNanos, "nowNanos"); + if (cooldownNanos <= 0) { + throw new IllegalArgumentException("cooldownNanos must be positive: " + cooldownNanos); + } + this.cooldownNanos = cooldownNanos; + } + + /** + * Inert quarantine — nothing is ever quarantined unless {@link #quarantine} is actually called on + * this instance. The explicit stand-in a caller (or a test not exercising this feature) passes + * instead of a defaulting overload, exactly like {@code ExhaustedPatternLookup.none()}. + */ + public static BackendQuarantine none() { + return new BackendQuarantine(() -> 0L, 1); + } + + /** + * Quarantine {@code credentialId} for the configured cooldown, starting now. A repeat call while + * already quarantined restarts the cooldown at full length — a fresh refusal is fresh evidence the + * account is still exhausted, not a reason to let an earlier, shorter wait stand. + */ + public void quarantine(String credentialId) { + Objects.requireNonNull(credentialId, "credentialId"); + quarantinedUntilNanos.put(credentialId, nowNanos.getAsLong() + cooldownNanos); + } + + /** Whether {@code credentialId} is quarantined right now. */ + public boolean isQuarantined(String credentialId) { + return remainingNanos(credentialId) > 0; + } + + /** Seconds left on {@code credentialId}'s quarantine, or empty when it is not quarantined. */ + public OptionalLong remainingSeconds(String credentialId) { + long remaining = remainingNanos(credentialId); + return remaining > 0 ? OptionalLong.of(toSecondsRoundedUp(remaining)) : OptionalLong.empty(); + } + + /** + * Every currently-quarantined credential id and its remaining seconds (CB-578 stage B fleet + * reporting) — expired entries are never included. Not pruned from the backing map here: it stays + * small (bounded by the number of distinct credentials ever exhausted) and a lazily-stale entry is + * harmless, since every read already checks the deadline. + */ + public Map activeRemainingSeconds() { + Map out = new LinkedHashMap<>(); + quarantinedUntilNanos.forEach((credentialId, deadline) -> { + long remaining = deadline - nowNanos.getAsLong(); + if (remaining > 0) { + out.put(credentialId, toSecondsRoundedUp(remaining)); + } + }); + return out; + } + + private long remainingNanos(String credentialId) { + Long deadline = quarantinedUntilNanos.get(credentialId); + return deadline == null ? 0L : deadline - nowNanos.getAsLong(); + } + + private static long toSecondsRoundedUp(long nanos) { + return (nanos + 999_999_999L) / 1_000_000_000L; + } +} diff --git a/bridged/src/main/java/dev/ltms/bridged/placement/FixedPlacementPolicy.java b/bridged/src/main/java/dev/ltms/bridged/placement/FixedPlacementPolicy.java index e0896ce..995ca15 100644 --- a/bridged/src/main/java/dev/ltms/bridged/placement/FixedPlacementPolicy.java +++ b/bridged/src/main/java/dev/ltms/bridged/placement/FixedPlacementPolicy.java @@ -4,18 +4,28 @@ package dev.ltms.bridged.placement; * Backward-compatible placement: an unqualified spawn always resolves to the configured default * profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps and * reachability so that a pre-existing config behaves identically after upgrade. + * + *

    Quarantine (CB-578 stage B) is the one exception: a quarantined default is a credential that + * just refused on a usage limit, not a transient capacity or reachability concern, so {@code fixed} + * steps to the first non-quarantined candidate instead of walking straight back onto it. A fleet + * where nothing is ever quarantined never exercises this path, so today's behaviour is unchanged. */ final class FixedPlacementPolicy implements PlacementPolicy { @Override public PlacementCandidate select(PlacementContext ctx) { String d = ctx.defaultProfile(); - if (d != null && !d.isBlank()) { + if (d != null && !d.isBlank() && !ctx.quarantined().contains(d)) { return new PlacementCandidate(d, null, 1.0f, null); } - if (!ctx.candidates().isEmpty()) { - PlacementCandidate first = ctx.candidates().getFirst(); - return new PlacementCandidate(first.profile(), null, first.weight(), first.maxLoad()); + for (PlacementCandidate c : ctx.candidates()) { + if (!ctx.quarantined().contains(c.profile())) { + return new PlacementCandidate(c.profile(), null, c.weight(), c.maxLoad()); + } + } + if (d != null && !d.isBlank()) { + throw new PlacementException("worker profile '" + d + "' is quarantined (backend " + + "exhausted) and no un-quarantined candidate is available"); } throw new PlacementException("no worker profiles configured"); } diff --git a/bridged/src/main/java/dev/ltms/bridged/placement/PlacementContext.java b/bridged/src/main/java/dev/ltms/bridged/placement/PlacementContext.java index 068f526..ad73c16 100644 --- a/bridged/src/main/java/dev/ltms/bridged/placement/PlacementContext.java +++ b/bridged/src/main/java/dev/ltms/bridged/placement/PlacementContext.java @@ -11,9 +11,13 @@ import java.util.function.Function; * @param candidates every configured candidate; the policy filters out those at cap or unreachable * @param liveCount current live worker count per profile (from the session registry) * @param unreachable profiles already known to have failed in this spawn attempt + * @param quarantined profiles whose credential is currently quarantined (CB-578 stage B) — a + * {@code BACKEND_EXHAUSTED} classification put it, or a profile it shares a + * credential with, on cooldown. Filtered the same way as {@code unreachable}. */ public record PlacementContext(String defaultProfile, List candidates, Function liveCount, - Set unreachable) { + Set unreachable, + Set quarantined) { } diff --git a/bridged/src/main/java/dev/ltms/bridged/placement/PlacementPolicyUtil.java b/bridged/src/main/java/dev/ltms/bridged/placement/PlacementPolicyUtil.java index 1e32f8b..66f0ca6 100644 --- a/bridged/src/main/java/dev/ltms/bridged/placement/PlacementPolicyUtil.java +++ b/bridged/src/main/java/dev/ltms/bridged/placement/PlacementPolicyUtil.java @@ -12,13 +12,13 @@ final class PlacementPolicyUtil { } /** - * Candidates that are not known-unreachable and have not reached their maxLoad. - * A {@code null} maxLoad means unlimited. + * Candidates that are not known-unreachable, not quarantined (CB-578 stage B), and have not + * reached their maxLoad. A {@code null} maxLoad means unlimited. */ static List available(PlacementContext ctx) { List out = new ArrayList<>(); for (PlacementCandidate c : ctx.candidates()) { - if (ctx.unreachable().contains(c.profile())) { + if (ctx.unreachable().contains(c.profile()) || ctx.quarantined().contains(c.profile())) { continue; } Integer cap = c.maxLoad(); @@ -35,14 +35,17 @@ final class PlacementPolicyUtil { /** * Build a clear exception describing why every candidate was dropped: all at capacity, - * all unreachable, or a mix. + * all unreachable, all quarantined, or a mix. */ static PlacementException emptyException(PlacementContext ctx) { int atCap = 0; int unreachable = 0; + int quarantined = 0; for (PlacementCandidate c : ctx.candidates()) { Integer cap = c.maxLoad(); - if (ctx.unreachable().contains(c.profile())) { + if (ctx.quarantined().contains(c.profile())) { + quarantined++; + } else if (ctx.unreachable().contains(c.profile())) { unreachable++; } else if (cap != null && ctx.liveCount().apply(c.profile()) >= cap) { atCap++; @@ -53,6 +56,9 @@ final class PlacementPolicyUtil { if (total == 0) { return new PlacementException("no worker profiles configured"); } + if (quarantined == total) { + return new PlacementException("all worker profiles are quarantined (backend exhausted)"); + } if (atCap == total) { return new PlacementException("all worker profiles are at maxLoad"); } @@ -60,6 +66,7 @@ final class PlacementPolicyUtil { return new PlacementException("all worker profiles are unreachable"); } return new PlacementException("no worker profile available: " + atCap + " at maxLoad, " - + unreachable + " unreachable, " + (total - atCap - unreachable) + " remaining"); + + unreachable + " unreachable, " + quarantined + " quarantined, " + + (total - atCap - unreachable - quarantined) + " remaining"); } } diff --git a/bridged/src/test/java/dev/ltms/bridged/config/ConfigRefTest.java b/bridged/src/test/java/dev/ltms/bridged/config/ConfigRefTest.java index 735cd96..7b33a21 100644 --- a/bridged/src/test/java/dev/ltms/bridged/config/ConfigRefTest.java +++ b/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, diff --git a/bridged/src/test/java/dev/ltms/bridged/inject/CompletionResolverTest.java b/bridged/src/test/java/dev/ltms/bridged/inject/CompletionResolverTest.java index a778b00..7c78236 100644 --- a/bridged/src/test/java/dev/ltms/bridged/inject/CompletionResolverTest.java +++ b/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 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 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)); diff --git a/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java b/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java index 7b1af8b..523137d 100644 --- a/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java +++ b/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; } diff --git a/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java b/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java index 43ba54d..7568244 100644 --- a/bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java +++ b/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 diff --git a/bridged/src/test/java/dev/ltms/bridged/member/CompositePeerLauncherTest.java b/bridged/src/test/java/dev/ltms/bridged/member/CompositePeerLauncherTest.java index b1f1994..cbc5078 100644 --- a/bridged/src/test/java/dev/ltms/bridged/member/CompositePeerLauncherTest.java +++ b/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 order-preserving 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 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 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 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 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))); + } } diff --git a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java index eb5de30..0f4fe23 100644 --- a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java +++ b/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); diff --git a/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java b/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java new file mode 100644 index 0000000..de10c0d --- /dev/null +++ b/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 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)); + } +} diff --git a/bridged/src/test/java/dev/ltms/bridged/placement/PlacementPolicyTest.java b/bridged/src/test/java/dev/ltms/bridged/placement/PlacementPolicyTest.java index 105056b..9c8a1c3 100644 --- a/bridged/src/test/java/dev/ltms/bridged/placement/PlacementPolicyTest.java +++ b/bridged/src/test/java/dev/ltms/bridged/placement/PlacementPolicyTest.java @@ -24,7 +24,13 @@ class PlacementPolicyTest { private static PlacementContext ctx(List candidates, Function liveCount, Set unreachable) { - return new PlacementContext("b", candidates, liveCount, unreachable); + return ctx(candidates, liveCount, unreachable, Set.of()); + } + + private static PlacementContext ctx(List candidates, + Function liveCount, + Set unreachable, Set quarantined) { + return new PlacementContext("b", candidates, liveCount, unreachable, quarantined); } private static PlacementContext ctx(List 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 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 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 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(); From 29ccb747ada4071e37fde1cd71b3000fe0d526cd Mon Sep 17 00:00:00 2001 From: Dai Ha Date: Sat, 15 Aug 2026 10:33:44 +0200 Subject: [PATCH 2/2] CB-578 stage B: make BackendQuarantine.none() actually inert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Added at merge review. none() held a clock frozen at 0 with a 1ns cooldown, so a quarantine() call on it recorded a deadline that could never pass — the credential would be locked out for the life of the daemon. Two production CompositePeerLauncher constructors default to none(), so that failure would have been silent and permanent. The implementer documented the limitation honestly rather than hiding it, but a stand-in named none() should not need the caveat. quarantine() is now a no-op on that instance, with a test asserting it. An inert value must omit the fact, never invent one. --- .../bridged/placement/BackendQuarantine.java | 22 ++++++++++++++++--- .../placement/BackendQuarantineTest.java | 19 ++++++++++++---- 2 files changed, 34 insertions(+), 7 deletions(-) diff --git a/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java b/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java index 1917a58..abdb95d 100644 --- a/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java +++ b/bridged/src/main/java/dev/ltms/bridged/placement/BackendQuarantine.java @@ -27,6 +27,8 @@ public final class BackendQuarantine { private final ConcurrentHashMap quarantinedUntilNanos = new ConcurrentHashMap<>(); private final LongSupplier nowNanos; private final long cooldownNanos; + /** True only for {@link #none()}. See {@link #quarantine} for why this exists. */ + private final boolean inert; /** * @param nowNanos monotonic clock, injected for testability @@ -34,29 +36,43 @@ public final class BackendQuarantine { * must be positive */ public BackendQuarantine(LongSupplier nowNanos, long cooldownNanos) { + this(nowNanos, cooldownNanos, false); + } + + private BackendQuarantine(LongSupplier nowNanos, long cooldownNanos, boolean inert) { this.nowNanos = Objects.requireNonNull(nowNanos, "nowNanos"); if (cooldownNanos <= 0) { throw new IllegalArgumentException("cooldownNanos must be positive: " + cooldownNanos); } this.cooldownNanos = cooldownNanos; + this.inert = inert; } /** - * Inert quarantine — nothing is ever quarantined unless {@link #quarantine} is actually called on - * this instance. The explicit stand-in a caller (or a test not exercising this feature) passes + * Inert quarantine — {@link #quarantine} does nothing on this instance, so nothing is ever + * quarantined. The explicit stand-in a caller (or a test not exercising this feature) passes * instead of a defaulting overload, exactly like {@code ExhaustedPatternLookup.none()}. */ public static BackendQuarantine none() { - return new BackendQuarantine(() -> 0L, 1); + return new BackendQuarantine(() -> 0L, 1, true); } /** * Quarantine {@code credentialId} for the configured cooldown, starting now. A repeat call while * already quarantined restarts the cooldown at full length — a fresh refusal is fresh evidence the * account is still exhausted, not a reason to let an earlier, shorter wait stand. + * + *

    On {@link #none()} this is a no-op. It has to be: that instance holds a clock frozen at 0, + * so recording a deadline would produce a quarantine that never expires — a credential locked out + * for the life of the daemon. Two production {@code CompositePeerLauncher} constructors default to + * {@code none()}, so the failure would be silent and permanent. An inert stand-in must omit the + * fact, never invent one. */ public void quarantine(String credentialId) { Objects.requireNonNull(credentialId, "credentialId"); + if (inert) { + return; + } quarantinedUntilNanos.put(credentialId, nowNanos.getAsLong() + cooldownNanos); } diff --git a/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java b/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java index de10c0d..5403f9e 100644 --- a/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java +++ b/bridged/src/test/java/dev/ltms/bridged/placement/BackendQuarantineTest.java @@ -90,16 +90,27 @@ class BackendQuarantineTest { @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 noneIgnoresAQuarantineCallInsteadOfLockingTheCredentialForever() { + // .none() holds a clock frozen at 0, so if #quarantine recorded a deadline the credential + // would never expire — locked out for the life of the daemon. Two production + // CompositePeerLauncher constructors default to none(), so that failure would be silent and + // permanent. An inert stand-in must omit the fact, never invent one. + BackendQuarantine q = BackendQuarantine.none(); + + q.quarantine("shared-openai"); + + assertFalse(q.isQuarantined("shared-openai"), "none() must not quarantine anything"); + assertTrue(q.remainingSeconds("shared-openai").isEmpty()); + assertTrue(q.activeRemainingSeconds().isEmpty()); + } + @Test void aNonPositiveCooldownIsRejected() { assertThrows(IllegalArgumentException.class, () -> new BackendQuarantine(() -> 0L, 0L));