Compare commits

..

1 Commits

Author SHA1 Message Date
Dai Ha d9e45f8c2a CB-566: add fleet charter config
CI / build (pull_request) Successful in 1m13s
CI / contract (pull_request) Successful in 1m15s
2026-08-15 04:49:53 +02:00
55 changed files with 600 additions and 3517 deletions
+14 -36
View File
@@ -47,18 +47,15 @@ bind:
# (say a Claude lead and an opencode lead) work as peers: the second is silently demoted and refused
# every orchestration call. List each lead's pane here and all of them resolve as leads.
#
# tab → the ONLY field identity depends on (CB-579); the exact label of the tab hosting the lead.
# Label the tab yourself, or let bridged label one it launches — see `fleet.leaders:` below.
# terminal → the ONLY field identity depends on; get it from that session's bridge_whoami
# kind/model → descriptive; they document what runs in the pane and are echoed by bridge_whoami
#
# A lead's tab must already carry its label (or be launched by bridged, which labels it) — there is
# no terminal id to paste in and nothing to re-pin when the session restarts: the tab survives, so
# the same label resolves the same lead again on the next scan.
# A lead is never spawned — it pre-exists, which is exactly why it must be named rather than created.
# `bridge_whoami` reports `{"role":"primary","leader":"<name>"}`; role stays "primary" because a lead
# IS a primary for authorization, so nothing that keys on the role breaks.
#
# KEEP `primary:` when adding leads: it still addresses the CB-307 push loop, which needs a single
# destination for its nudges, and is a separate mechanism from lead identity — see `fleet.leaders:`.
# destination for its nudges. If both name the same terminal, the `fleet.leaders:` entry wins.
#
# Leads are configured under `fleet.leaders:` — see THE FLEET further down.
#
@@ -88,16 +85,6 @@ bind:
# backoffMs: 60000
# quietNudgeCap: 3
# Fleet health detection is dormant unless enabled. It reads one whole-fleet agent list per tick.
# It can run without a webhook; bridge_list then reports healthCoverage: detection-only.
# health:
# enabled: true
# intervalSeconds: 30 # minimum 15
# workingSuspectAfterSeconds: 600 # minimum 300
# paneProbeIntervalSeconds: 60 # minimum 60
# notifications:
# mode: disabled # disabled (default) or webhook
# herdr Unix socket. Omit to use the client default
# (${HERDR_SOCKET_PATH:-~/.config/herdr/herdr.sock}).
herdrSocket: ~/.config/herdr/herdr.sock
@@ -138,12 +125,6 @@ herdrSocket: ~/.config/herdr/herdr.sock
# SSH is unaffected). The token value itself is never stored in this file.
# gitHostEnv → host env var holding the forge host (default GITEA_HOST). Injected as
# GITEA_HOST *only* alongside a resolved gitTokenEnv.
# exhaustedPattern → regex matched against a completion-fallback scrape (CB-578 stage A) to
# classify a turn that ended with no bridge_reply as the backend having
# refused on a subscription usage limit, rather than a real answer. Opt-in —
# 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.
# env → extra environment for this profile's workers, as a literal key/value map
# (CB-511). Use it to give workers a toolchain.
#
@@ -177,7 +158,6 @@ profiles:
maxLoad: 2 # max live workers on this profile (omit for unlimited)
# 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)
# 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
@@ -319,18 +299,15 @@ fleet:
# Panes that orchestrate rather than are orchestrated. A lead may now be CREATED as well as
# recognised: give it a `profile:` and the daemon launches the shortfall when fewer than
# `instances` are live. Omit `profile:` and it is recognise-only, as before.
# `instances` are live. Give it only a `terminal:` and it is recognise-only, as before.
#
# `tab:` (CB-579) is REQUIRED and is the only field identity depends on — the exact label of the
# tab hosting the lead, matched case-insensitively. Label the tab yourself and put that same
# string here, and the pane is recognised on the next rescan. Reopen the tab later, or the session
# inside it restarts — the terminal id changes; the tab, and its label, do not, so no config edit
# follows a restart.
# `tabPrefix` is the naming convention that finds a lead without pasting a terminal id: label the
# tab `lead: <name>` when you open it and the pane is recognised on the next rescan. Reopen the
# tab later and the id changes; the label does not.
#
# A lead the daemon launches is labelled BY the daemon with this same `tab:` value, so it is found
# A lead the daemon launches is labelled BY the daemon, using the same convention, so it is found
# by the same scan. A lead counts as live only when herdr also reports a running agent in that
# tab — a label left behind by a session that died does not block the relaunch, and a tab that is
# gone entirely drops out of the next scan rather than being remembered forever.
# tab — a label left behind by a session that died does not block the relaunch.
#
# An auto-launched lead is NOT a member: it gets no worker reply charter, is never registered with
# the session lifecycle (the idle reaper would kill your orchestrator), and stays on the
@@ -339,9 +316,10 @@ fleet:
# opus-5.0:
# profile: opus # omit to never create this lead, only recognise it
# instances: 1 # desired live count; only the shortfall is launched. 0 = off
# tab: "lead: opus-5.0" # REQUIRED — the exact tab label this lead lives in
# tabPrefix: "lead:" # only used to guard against a worker tabLabel colliding with
# # this convention at startup; plays no part in matching a lead
# terminal: term_0123456789abcd # optional hand-pin; usually found by tabPrefix instead.
# # A running agent on this terminal also counts as live, so a
# # lead you opened by hand is not relaunched under you.
# tabPrefix: "lead:" # `lead: opus-5.0` ⇒ a lead named opus-5.0 (case-insensitive)
# scanIntervalSeconds: 10 # rescan cadence, and the worst case before a new tab is seen
# workspace: leads # where a launched lead's tab is created (default "leads").
# # MUST NOT be a member workspace — those are excluded from the
@@ -349,7 +327,7 @@ fleet:
# cwd: /path/to/repo # the launched lead's working directory (default: bridged's own)
# kind: claude # descriptive; reported by bridge_whoami
# gpt-sol-5.6:
# tab: "lead: gpt-sol-5.6"
# terminal: term_fedcba9876543
# kind: opencode
# model: openai/gpt-5.6-terra
@@ -13,7 +13,6 @@ import dev.ltms.bridged.herdr.PaneLocator;
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.Injector;
import dev.ltms.bridged.inject.StatusPoller;
import dev.ltms.bridged.inject.TurnListener;
@@ -24,7 +23,6 @@ import dev.ltms.bridged.mcp.BridgeMcp;
import dev.ltms.bridged.mcp.ConnectionIdentity;
import dev.ltms.bridged.metrics.BridgedMetrics;
import dev.ltms.bridged.metrics.Metrics;
import dev.ltms.bridged.health.FleetHealthMonitor;
import dev.ltms.bridged.mcp.PrimaryRegistry;
import dev.ltms.bridged.mcp.LsofPeerPidLookup;
import dev.ltms.bridged.mcp.LsofProcessCwdLookup;
@@ -61,7 +59,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
@@ -132,13 +129,13 @@ public final class Bridged {
adapters.add(new ClaudeCodeLauncher(agents, spaces, guard,
claudeProfiles, cfg.effectiveDefaultProfile(), System::getenv,
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs(),
() -> config.get().fleet()));
() -> config.get().fleet().tabLabel()));
}
if (!opencodeProfiles.isEmpty()) {
adapters.add(new OpenCodeLauncher(agents, spaces,
opencodeProfiles, cfg.effectiveDefaultProfile(), System::getenv,
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs(),
() -> config.get().fleet()));
() -> config.get().fleet().tabLabel()));
}
AtomicReference<Function<String, Integer>> liveCountRef = new AtomicReference<>(_ -> 0);
PeerLauncher workers = new CompositePeerLauncher(
@@ -195,34 +192,34 @@ public final class Bridged {
if (leadTerminals.size() > 1) {
log.info("leads: {} panes recognised {}", leadTerminals.size(), leadTerminals.values());
}
// CB-531: on top of the legacy primary.terminal pin, discover leads by the tab labels the
// operator writes. CB-557 moved the settings onto the lead they describe, so scanning is on
// whenever a `fleet.leaders:` entry exists — with no leads configured the supplier is a
// constant and never touches herdr, exactly as a missing `leadScan:` block used to behave.
// CB-579: each lead now names its own exact `tab:` label, so one scanner discovers every
// configured lead regardless of how differently their tabs are labelled — the old
// single-shared-tabPrefix limitation (and its warning) is gone.
// CB-531: on top of the static registry, discover leads by the tab labels the operator
// writes. CB-557 moved the settings onto the lead they describe, so scanning is on whenever
// a `fleet.leaders:` entry exists — with no leads configured the supplier is a constant and
// never touches herdr, exactly as a missing `leadScan:` block used to behave.
final Supplier<Map<String, String>> leads;
var leaders = cfg.fleet().leaders();
if (!leaders.isEmpty()) {
// One scanner, so one prefix and one interval. Distinct per-lead prefixes would need a
// scanner each; until a config actually wants that, take the first entry's settings and
// say so, rather than silently honouring one lead's prefix and dropping another's.
var scan = leaders.values().iterator().next();
Set<String> memberSpaces = cfg.profiles().values().stream()
.map(BridgedConfig.Profile::workspace)
.filter(Objects::nonNull)
.collect(Collectors.toSet());
Map<String, String> tabToName = new LinkedHashMap<>();
leaders.forEach((name, leader) -> {
if (leader != null && leader.tab() != null && !leader.tab().isBlank()) {
tabToName.put(leader.tab(), name);
}
});
// One shared rescan cadence: still taken from the first entry, as before — it is an
// operational cadence, not identity, so there is no correctness reason to give every
// lead its own scanner.
int scanIntervalSeconds = leaders.values().iterator().next().scanIntervalSeconds();
leads = new LeadTabScanner(herdr, tabToName, memberSpaces,
TimeUnit.SECONDS.toNanos(scanIntervalSeconds), System::nanoTime);
log.info("lead scan: tabs {} host a lead (rescan every {}s, member spaces {} excluded)",
tabToName.keySet(), scanIntervalSeconds, memberSpaces);
leads = new LeadTabScanner(herdr, scan.tabPrefix(), memberSpaces, leadTerminals,
TimeUnit.SECONDS.toNanos(scan.scanIntervalSeconds()), System::nanoTime);
log.info("lead scan: tabs labelled '{}…' host a lead (rescan every {}s, member spaces {} "
+ "excluded)",
scan.tabPrefix(), scan.scanIntervalSeconds(), memberSpaces);
long distinctPrefixes = leaders.values().stream()
.map(BridgedConfig.Leader::tabPrefix).distinct().count();
if (distinctPrefixes > 1) {
log.warn("fleet.leaders declares {} different tabPrefix values; only '{}' is scanned "
+ "for. Give every lead the same tabPrefix, or leads under the others "
+ "will not be discovered.",
distinctPrefixes, scan.tabPrefix());
}
} else {
leads = () -> leadTerminals;
}
@@ -255,24 +252,7 @@ public final class Bridged {
// The blocking message endpoint (CB-104) is the producer; the poller is inert until then.
// CB-106: a confirmed turn completion resolves a blocked send whose worker never replied.
Rendezvous rendezvous = new Rendezvous();
// CB-578 stage A: classify a completion-fallback scrape that matches a profile's configured
// usage-limit refusal as BACKEND_EXHAUSTED rather than handing it back as a real answer.
// Compiled once at startup, keyed by profile name; a profile with no exhaustedPattern is
// simply absent here, so its workers keep today's completion-fallback behaviour unchanged.
Map<String, Pattern> exhaustedPatternsByProfile = new LinkedHashMap<>();
cfg.profiles().forEach((name, profile) -> {
if (profile.hasExhaustedPattern()) {
exhaustedPatternsByProfile.put(name, Pattern.compile(profile.exhaustedPattern()));
}
});
ExhaustedPatternLookup exhaustedPatterns = target -> sessions.roster().stream()
.filter(session -> target.equals(session.terminalId()))
.findFirst()
.map(session -> exhaustedPatternsByProfile.get(session.profile()))
.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);
CompletionResolver completion = new CompletionResolver(agents, rendezvous);
// 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();
@@ -295,9 +275,9 @@ public final class Bridged {
}
@Override
public void onDelivered(String target, dev.ltms.bridged.msg.TurnToken token) {
completion.onDelivered(target, token);
sessions.onDelivered(target, token);
public void onDelivered(String target) {
completion.onDelivered(target);
sessions.onDelivered(target);
}
@Override
@@ -305,12 +285,6 @@ public final class Bridged {
completion.onTurnFailed(target);
sessions.onTurnFailed(target);
}
@Override
public void onTurnFailed(String target, String reason) {
completion.onTurnFailed(target, reason);
sessions.onTurnFailed(target);
}
};
Injector injector = new Injector(agents, turnListener, deliverableTo(presence, leads),
presence::forget);
@@ -374,28 +348,6 @@ public final class Bridged {
MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
pushLoop, metrics);
// Health is a slow whole-fleet observer. Keep it separate from the 250ms delivery poller.
final FleetHealthMonitor healthMonitor;
var healthScheduler = Executors.newSingleThreadScheduledExecutor(r ->
Thread.ofVirtual().name("bridge-health-").unstarted(r));
if (cfg.health() != null && cfg.health().isEnabled()) {
// CB-580: a member found GONE/NEVER_READY must fail whatever ticket is waiting on it,
// through the same idempotent target-wide operation CB-516 already uses on release.
healthMonitor = new FleetHealthMonitor(agents, sessions::roster, messages, healthScheduler,
System::nanoTime, cfg.health().intervalOrDefault(), messages::abandon);
String coverage = FleetHealthMonitor.coverage(true,
cfg.health().notifications() != null && cfg.health().notifications().configured());
if ("detection-only".equals(coverage)) {
log.warn("fleet health: {} (no notification sink configured)", coverage);
} else {
log.info("fleet health: {}", coverage);
}
healthMonitor.start();
} else {
healthMonitor = null;
healthScheduler.shutdownNow();
}
// CB-520: the reply inbox only consumes for agents this gateway owns. own on acquire,
// release on teardown. Do this before CB-516 so the inbox is owned before any reply can land.
sessions.onAcquire(replyInbox::own);
@@ -431,16 +383,7 @@ public final class Bridged {
}
BridgeMcp mcp = new BridgeMcp(messages, workers, sessions, identity, presence,
primaryRegistry, callers, metrics, new BridgeMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.maxLoad();
}, () -> config.get().profiles().keySet(), System::nanoTime),
new BridgeMcp.HealthCoverageSource(() -> {
var health = config.get().health();
return FleetHealthMonitor.coverage(health != null && health.isEnabled(),
health != null && health.notifications() != null && health.notifications().configured());
}));
primaryRegistry, callers, metrics);
// 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.
@@ -461,7 +404,6 @@ public final class Bridged {
messages.close();
pushLoop.close();
if (heartbeat != null) heartbeat.close(); // CB-551: stop the idle-lead heartbeat scheduler
if (healthMonitor != null) healthMonitor.stop();
if (configWatcher != null) configWatcher.stop(); // CB-559: stop polling the config file
mcp.close();
if (reaper != null) reaper.stop();
@@ -73,7 +73,6 @@ public record BridgedConfig(
Primary primary,
Fleet fleet,
LeadHeartbeat leadHeartbeat,
Health health,
String placement,
Auth auth,
ConfigReload configReload) {
@@ -84,16 +83,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);
}
/** Back-compat form before the optional {@code health:} block was added. */
public BridgedConfig(Bind bind, String herdrSocket, Map<String, Profile> profiles, Guard guard,
String worktreeRoot, Lifecycle lifecycle, Integer spawnReadyTimeoutMs,
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, placement, auth, null);
}
/**
@@ -181,12 +171,6 @@ public record BridgedConfig(
* opposite intents). For the same reason, an {@code env:} entry naming
* {@code ANTHROPIC_BASE_URL} or {@code ANTHROPIC_AUTH_TOKEN} is refused at
* config load (CB-542): on the subscription path no guard would vet it.
* @param exhaustedPattern regex matched against a completion-fallback scrape (CB-578 stage A) to
* classify a turn that ended with no {@code bridge_reply} as the backend
* having refused on a subscription usage limit, rather than a real answer.
* {@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.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
public record Profile(String profile, String baseUrl, String model,
@@ -199,8 +183,7 @@ public record BridgedConfig(
Map<String, String> env,
Float weight,
Integer maxLoad,
Boolean subscription,
String exhaustedPattern) {
Boolean subscription) {
/** Peer kind spawned by {@link dev.ltms.bridged.member.ClaudeCodeLauncher} (the default). */
public static final String KIND_CLAUDE_CODE = "claude-code";
@@ -243,9 +226,6 @@ public record BridgedConfig(
weight = (weight == null || weight <= 0.0f) ? 1.0f : weight;
maxLoad = (maxLoad == null || maxLoad <= 0) ? null : maxLoad;
subscription = (subscription != null && subscription) ? Boolean.TRUE : Boolean.FALSE;
// 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;
}
/**
@@ -258,7 +238,7 @@ public record BridgedConfig(
String placement, String workspace, String tabLabel, String mcpUrl,
String cwd, List<String> parityOverlay) {
this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
mcpUrl, cwd, parityOverlay, null, null, null, null, null, null, null, null);
mcpUrl, cwd, parityOverlay, null, null, null, null, null, null, null);
}
/**
@@ -270,7 +250,7 @@ public record BridgedConfig(
String placement, String workspace, String tabLabel, String mcpUrl,
String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv) {
this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, null, null, null, null, null, null);
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, null, null, null, null, null);
}
/**
@@ -283,14 +263,13 @@ public record BridgedConfig(
String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv,
String kind) {
this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, null, null, null, null, null);
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, null, null, null, null);
}
/** A copy with {@code profile} set — used to default a profile to its {@code workers} key. */
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);
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, subscription);
}
/** True when this profile is served by the Claude Code adapter (the default kind). */
@@ -323,12 +302,7 @@ public record BridgedConfig(
String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv,
String kind, Map<String, String> env, Float weight, Integer maxLoad) {
this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, null, null);
}
/** True when this profile's CB-578 stage A backend-exhausted classification is configured. */
public boolean hasExhaustedPattern() {
return exhaustedPattern != null;
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad, null);
}
/** True when this profile's workers are granted a forge token to open their own PR (CB-302). */
@@ -403,17 +377,6 @@ public record BridgedConfig(
boolean clearAfterTurn) {
}
/** Optional fleet detection. A missing block stays dormant. */
@JsonIgnoreProperties(ignoreUnknown = true)
public record Health(Boolean enabled, Integer intervalSeconds, Integer workingSuspectAfterSeconds,
Integer paneProbeIntervalSeconds, Notifications notifications) {
public boolean isEnabled() { return Boolean.TRUE.equals(enabled); }
public int intervalOrDefault() { return Math.max(15, intervalSeconds == null ? 30 : intervalSeconds); }
public record Notifications(String mode) {
public boolean configured() { return "webhook".equalsIgnoreCase(mode); }
}
}
/**
* External AMQP broker for durable, cross-restart reply delivery (CB-307 Stage 2). Its mere
* presence swaps the in-memory {@code ReplyInbox} for the AMQP-backed adapter; absent, bridged
@@ -474,32 +437,24 @@ public record BridgedConfig(
* pre-existed, which is why it had to be recognised by configuration rather than created. With
* {@code profile} and {@code instances} the daemon may stand one up when none is live, so the
* pane no longer has to exist before the daemon does. Recognition still comes first: a lead
* already running in its configured {@code tab} is adopted, and only the shortfall is launched.
*
* <p><b>{@code tab} replaced {@code terminal} (CB-579).</b> A herdr {@code terminal_id} changes
* every time the lead's session restarts, so pinning one cost a config edit and a daemon restart
* per restart. A tab is stable: a human opens it once, it holds exactly one pane, and its label
* survives restarts of the agent inside it — so identity is now the tab label alone.
* already running under {@code tabPrefix} is adopted, and only the shortfall is launched.
*
* @param profile the {@code profiles:} entry to launch this lead on when one must
* be created; {@code null} ⇒ recognise-only, never create
* @param tab the exact tab label hosting this lead, matched case-insensitively;
* the only field identity depends on. Required — a lead with no
* {@code tab} can never be discovered, launched or not
* @param terminal the lead's herdr {@code terminal_id} when pinned by hand; the only
* field identity depends on. {@code null} ⇒ found by {@code tabPrefix}
* @param instances how many of this lead should be live (default 1). The daemon
* launches only the shortfall, so a restart adopts rather than doubles
* @param tabPrefix no longer used to find a lead's tab — {@code tab} is matched
* exactly. Its only remaining job is the startup collision guard
* ({@link #validateLeadTabPrefixes()}), which still uses it to refuse
* a worker {@code tabLabel} template that could be misread as a lead.
* Default {@code "lead:"}
* @param tabPrefix label prefix marking this lead's tab, matched case-insensitively;
* the remainder is the lead's name ({@code "lead: opus"} →
* {@code opus}). Default {@code "lead:"}
* @param scanIntervalSeconds how long a tab scan is cached before herdr is asked again; also the
* worst case before a newly-labelled tab is recognised. Default 10
* @param kind which agent runs there ({@code claude}, {@code opencode}, …)
* @param model the model or selector it runs, for operators reading the roster
*/
@JsonIgnoreProperties(ignoreUnknown = true)
public record Leader(String profile, String tab, Integer instances, String tabPrefix,
public record Leader(String profile, String terminal, Integer instances, String tabPrefix,
Integer scanIntervalSeconds, String kind, String model,
String workspace, String cwd) {
@@ -517,13 +472,12 @@ public record BridgedConfig(
(scanIntervalSeconds == null || scanIntervalSeconds <= 0) ? 10 : scanIntervalSeconds;
workspace = (workspace == null || workspace.isBlank())
? DEFAULT_WORKSPACE : workspace.strip();
tab = (tab == null || tab.isBlank()) ? null : tab.strip();
}
/** Back-compat 7-arg form — no workspace or cwd, so both take their defaults. */
public Leader(String profile, String tab, Integer instances, String tabPrefix,
public Leader(String profile, String terminal, Integer instances, String tabPrefix,
Integer scanIntervalSeconds, String kind, String model) {
this(profile, tab, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
this(profile, terminal, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
}
/** True when this lead may be launched by the daemon rather than only recognised. */
@@ -531,9 +485,9 @@ public record BridgedConfig(
return profile != null && !profile.isBlank() && instances > 0;
}
/** The tab label an auto-launched instance of this lead gets — its configured {@code tab}. */
public String tabLabel() {
return tab;
/** The tab label an auto-launched instance of this lead gets — what the scanner reads back. */
public String tabLabel(String name) {
return tabPrefix + " " + name;
}
}
@@ -608,15 +562,7 @@ public record BridgedConfig(
tabLabel = (tabLabel == null || tabLabel.isBlank()) ? DEFAULT_TAB_LABEL : tabLabel;
}
/**
* A fleet with no configured launch charters — the shape every deployment had before
* CB-566, and what most tests want.
*
* <p>Kept deliberately, even though an overload that drops a new field is normally the
* shape to avoid. It is safe here because nothing <em>reads</em> a charter through a
* constructor: the launcher reads {@code fleet.charters()} from the live config. Jackson
* binds the canonical constructor, so this one cannot swallow an operator's YAML.
*/
/** Convenience constructor for code that does not configure launch charters. */
public Fleet(Map<String, Leader> leaders, Map<String, Slot> architects,
Map<String, Slot> developers, Map<String, Slot> reviewers, String tabLabel) {
this(leaders, architects, developers, reviewers, null, tabLabel);
@@ -732,20 +678,28 @@ public record BridgedConfig(
}
/**
* The terminal → lead-name map seeded from the legacy singular {@code primary:} pin (CB-530).
* The terminal → lead-name map that {@link dev.ltms.bridged.auth.CallerResolver} resolves
* against, merging the {@code leaders:} registry with the legacy singular {@code primary:} pin.
*
* <p>{@code fleet.leaders} no longer carries a per-entry terminal pin (CB-579): a lead's identity
* comes from its {@code tab} alone, resolved live by {@code LeadTabScanner}. This method now
* exists only for the {@code primary.terminal} fallback — a config that never migrated off it
* still resolves that one pane as a lead named {@code "primary"}, exactly as before CB-530.
* <p>Precedence: an explicit {@code leaders:} entry wins over the {@code primary:} pin for the
* same terminal. The pin is the older, less expressive spelling of the same fact, so when both
* name a pane the named entry is the one an operator meant. The pin is still honoured on its
* own — a config carrying only {@code primary:} behaves exactly as it did before CB-530.
*
* @return an unmodifiable map, empty when {@code primary.terminal} is not configured (nothing is
* pinned, and every pane therefore resolves as a worker — the pre-CB-307 behaviour)
* @return an unmodifiable map, empty when neither block is configured (nothing is pinned, and
* every pane therefore resolves as a worker — the pre-CB-307 behaviour)
*/
public Map<String, String> leaderTerminals() {
Map<String, String> byTerminal = new LinkedHashMap<>();
if (fleet != null) {
fleet.leaders().forEach((name, leader) -> {
if (leader != null && leader.terminal() != null && !leader.terminal().isBlank()) {
byTerminal.put(leader.terminal(), name);
}
});
}
if (primary != null && primary.terminal() != null && !primary.terminal().isBlank()) {
byTerminal.put(primary.terminal(), "primary");
byTerminal.putIfAbsent(primary.terminal(), "primary");
}
return Collections.unmodifiableMap(byTerminal);
}
@@ -850,14 +804,13 @@ public record BridgedConfig(
private static final Set<String> KNOWN_TOP_LEVEL_KEYS = Set.of(
"bind", "herdrSocket", "profiles", "guard", "worktreeRoot",
"lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs", "broker", "primary", "fleet",
"leadHeartbeat", "health", "placement", "auth", "configReload");
"leadHeartbeat", "placement", "auth", "configReload");
/** Load and validate config from {@code path}. */
public static BridgedConfig load(Path path) {
try {
String yaml = Files.readString(path);
rejectRenamedTopLevelKeys(yaml);
rejectLeaderTerminalKey(yaml);
warnUnknownTopLevelKeys(yaml, path);
rejectDuplicateMemberSlots(yaml);
BridgedConfig cfg = YAML.readValue(yaml, BridgedConfig.class);
@@ -1082,43 +1035,6 @@ public record BridgedConfig(
}
}
/**
* Reject a config whose {@code fleet.leaders.<name>} still carries the retired {@code terminal:}
* pin (CB-579), naming {@code tab:} as its replacement.
*
* <p>{@code Leader} is {@code @JsonIgnoreProperties(ignoreUnknown = true)}, so simply dropping
* the record component would make a leftover {@code terminal:} key silently no-op — the daemon
* would start, the pin would never take effect, and nothing would say why. Fatal and specific
* instead, exactly like {@link #rejectRenamedTopLevelKeys}, which this mirrors for a key one
* level deeper than the ones that method covers.
*
* @param yaml the raw config text
* @throws IllegalStateException when any {@code fleet.leaders.<name>.terminal} key is present
*/
static void rejectLeaderTerminalKey(String yaml) {
Map<?, ?> raw;
try {
raw = YAML.readValue(yaml, Map.class);
} catch (IOException | IllegalArgumentException e) {
return; // a malformed file is reported by the real parse, not here
}
if (raw == null || !(raw.get("fleet") instanceof Map<?, ?> fleet)
|| !(fleet.get("leaders") instanceof Map<?, ?> leaders)) {
return;
}
List<String> bad = leaders.entrySet().stream()
.filter(e -> e.getValue() instanceof Map<?, ?> leader && leader.containsKey("terminal"))
.map(e -> String.valueOf(e.getKey()))
.sorted()
.toList();
if (!bad.isEmpty()) {
throw new IllegalStateException("refusing to start: fleet.leaders entries ["
+ String.join(", ", bad) + "] still use the retired 'terminal:' key — replace it "
+ "with 'tab:', the exact tab label hosting the lead. A terminal_id changes on "
+ "every restart of the lead's session; a tab label does not.");
}
}
static List<String> unknownTopLevelKeys(String yaml) {
Map<?, ?> raw;
try {
@@ -1158,7 +1074,7 @@ public record BridgedConfig(
// 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.
return new BridgedConfig(b, herdrSocket, profiles, g, worktreeRoot, l, timeout, pollMs,
broker, primary, f, leadHeartbeat, health, placementOrDefault, a, configReload);
broker, primary, f, leadHeartbeat, placementOrDefault, a, configReload);
}
/**
@@ -1301,7 +1217,7 @@ public record BridgedConfig(
if (fleet == null || fleet.charters().isEmpty()) {
return;
}
List<String> valid = Arrays.stream(MemberRole.values())
List<String> valid = java.util.Arrays.stream(MemberRole.values())
.map(MemberRole::wireName)
.toList();
List<String> bad = new ArrayList<>();
@@ -1368,9 +1284,10 @@ public record BridgedConfig(
+ "', which is not a configured profiles: entry (have: " + profiles.keySet()
+ ").");
}
if (leader.tab() == null || leader.tab().isBlank()) {
bad.add("fleet.leaders." + name + " has no tab: — a lead is now found (and, if "
+ "auto-launched, labelled) purely by its tab, so every entry must name one.");
if (!leader.isCreatable() && (leader.terminal() == null || leader.terminal().isBlank())) {
bad.add("fleet.leaders." + name + " can neither be found nor created — it pins no "
+ "terminal: and names no profile: to launch one on. Give it one or the "
+ "other, or drop the entry.");
}
});
if (!bad.isEmpty()) {
@@ -28,13 +28,9 @@ import java.util.function.Supplier;
* <ul>
* <li><strong>Hot</strong> — re-read per use, so a reload takes effect on the next spawn:
* {@code fleet:} (every role pool, {@code charters}, and {@code tabLabel}),
* {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Those
* three are read through a supplier on {@code CompositePeerLauncher}, which is what makes
* them hot — not the fact that they are config. <strong>This does NOT include
* {@code fleet.leaders}</strong>: {@code Bridged.main} reads {@code cfg.fleet().leaders()}
* once at startup to build the {@code LeadTabScanner} and the {@code LeadLauncher}, and
* neither is reconstructed on reload — so a lead added, removed, or re-{@code tab}'d under
* {@code fleet.leaders} needs a restart, the same as any deferred key below.</li>
* {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Those three are read through a supplier on
* {@code CompositePeerLauncher}, which is what makes them hot — not the fact that they are
* config.</li>
* <li><strong>Deferred</strong> — 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:},
@@ -1,40 +0,0 @@
package dev.ltms.bridged.health;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.session.MemberSession;
/**
* Pure classifier. Collection and repair are deliberately outside this package.
* {@link HealthState#ERROR_ON_SCREEN} is not decided yet because it needs a bounded pane detection
* read and an adapter-specific fatal signature; status facts alone must not guess it.
*/
public final class FleetHealth {
private FleetHealth() { }
public static HealthDecision decide(HealthSnapshot s, HealthPrior prior, long nowNanos) {
if (s.controlLinkDown()) return result(HealthState.CONTROL_LINK_DOWN, false);
if (s.targetNotFound()) return result(HealthState.GONE, false);
if (s.sessionState() == MemberSession.State.SPAWNING && !s.present() && s.readinessGraceElapsed()) {
return result(HealthState.NEVER_READY, false);
}
if (s.orphanedDelegation()) return result(HealthState.DELEGATION_ORPHANED, false);
boolean disagreement = s.sessionState() == MemberSession.State.BUSY && s.acceptedDelivery()
&& (s.liveStatus() == AgentStatus.IDLE || s.liveStatus() == AgentStatus.DONE);
if (disagreement && prior.busyButDone()) return result(HealthState.TURN_BOUNDARY_LOST, true);
if (s.stalled()) return result(HealthState.STALL_SUSPECTED, disagreement);
if (s.replyStranded()) return result(HealthState.REPLY_STRANDED, disagreement);
if (s.queuedDelivery() || s.inboxMessage()) return result(HealthState.WORK_PENDING, disagreement);
if (s.sessionState() == MemberSession.State.SPAWNING) return result(HealthState.STARTING, disagreement);
if (s.acceptedDelivery() && s.liveStatus() == AgentStatus.BLOCKED) {
return result(HealthState.BLOCKED_AMBIGUOUS, disagreement);
}
// An accepted delivery remains bridge work even when herdr is late, unknown, or has already
// reported DONE once. It cannot be IDLE until the delegation has resolved.
if (s.acceptedDelivery()) return result(HealthState.WORKING, disagreement);
return result(HealthState.IDLE, disagreement);
}
private static HealthDecision result(HealthState state, boolean disagreement) {
return new HealthDecision(state, new HealthPrior(disagreement));
}
}
@@ -1,151 +0,0 @@
package dev.ltms.bridged.health;
import dev.ltms.bridged.herdr.Agent;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.msg.MessageService;
import dev.ltms.bridged.session.MemberSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
/** Slow whole-fleet evidence collection. It is deliberately separate from the delivery poller. */
public final class FleetHealthMonitor {
private static final Logger log = LoggerFactory.getLogger(FleetHealthMonitor.class);
/** Bounded attempts to run {@link #failTarget} for one transition. Never retried tick-to-tick (CB-580). */
static final int MAX_FAIL_TARGET_ATTEMPTS = 3;
private final AgentControl agents;
private final Supplier<List<MemberSession>> roster;
private final MessageService messages;
private final ScheduledExecutorService scheduler;
private final LongSupplier clock;
private final long intervalSeconds;
private final BiConsumer<String, String> failTarget;
private final Map<String, HealthPrior> priors = new HashMap<>();
private final Map<String, HealthState> states = new HashMap<>();
// These facts need the evidence publishers introduced by later M4 units. They are not negatives.
private static final boolean NOT_YET_OBSERVED = false;
/**
* @param failTarget CB-568's idempotent target-wide failure operation (e.g. {@code messages::abandon}),
* invoked once when a member transitions into a terminal health state. Required —
* there is deliberately no defaulting overload; a caller that does not want the
* fail-tickets-on-terminal-health behavior must pass an explicit inert value (see
* {@code TestTurnTokens.inert} / {@code BridgeMcp.CapacitySource.none()} for the pattern).
*/
public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds,
BiConsumer<String, String> failTarget) {
this.agents = agents;
this.roster = roster;
this.messages = messages;
this.scheduler = scheduler;
this.clock = clock;
this.intervalSeconds = intervalSeconds;
this.failTarget = Objects.requireNonNull(failTarget, "failTarget");
}
/** Pure per-member decision seam. */
static HealthDecision decide(HealthSnapshot snapshot, HealthPrior prior, long nowNanos) {
return FleetHealth.decide(snapshot, prior, nowNanos);
}
public void start() { scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS); }
public void stop() { scheduler.shutdownNow(); }
// Package-private so tests can run one tick without waiting.
void tick() {
try {
List<Agent> agentsNow = agents.list(); // Exactly one list call for this complete observation.
List<MemberSession> rosterNow = roster.get(); // One in-memory roster snapshot for this tick.
Map<String, Agent> live = new HashMap<>();
for (Agent agent : agentsNow) live.put(agent.terminalId(), agent);
HashSet<String> current = new HashSet<>();
for (MemberSession session : rosterNow) {
current.add(session.terminalId());
Agent agent = live.get(session.terminalId());
AgentStatus status = agent == null ? AgentStatus.UNKNOWN : agent.status();
boolean accepted = messages.hasAcceptedDelivery(session.terminalId());
HealthSnapshot snapshot = new HealthSnapshot(session.state(), status, accepted, NOT_YET_OBSERVED,
messages.hasInboxMessage(session.terminalId()), agent != null, NOT_YET_OBSERVED,
NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED);
HealthDecision decision = decide(snapshot, priors.getOrDefault(session.terminalId(), HealthPrior.NONE),
clock.getAsLong());
priors.put(session.terminalId(), decision.prior());
reportTransition(session.terminalId(), decision.state());
}
priors.keySet().retainAll(current);
states.keySet().retainAll(current);
} catch (Throwable error) {
// A list failure is health evidence, and must never kill the monitor's only scheduler task.
log.warn("fleet health collection failed; will retry next tick", error);
} finally {
if (!scheduler.isShutdown()) {
scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS);
}
}
}
void reportTransition(String target, HealthState next) {
HealthState previous = states.put(target, next);
if (previous == next) return;
if (fault(next)) {
log.warn("fleet health member={} state={} previous={}", target, next, previous);
} else if (previous != null && fault(previous)) {
log.info("fleet health member={} recovered state={} previous={}", target, next, previous);
}
// CB-580: a member entering GONE/NEVER_READY must not leave its waiting tickets pending
// forever. Fire exactly once per transition — never on a tick where the state is unchanged,
// which is what made the rejected commit call abandon() once per tick for as long as a
// member stayed terminal.
if (terminal(next)) {
failTerminalTarget(target, next);
}
}
private void failTerminalTarget(String target, HealthState state) {
String reason = "fleet health: member reached terminal state " + state.name();
RuntimeException last = null;
for (int attempt = 1; attempt <= MAX_FAIL_TARGET_ATTEMPTS; attempt++) {
try {
failTarget.accept(target, reason);
return;
} catch (RuntimeException error) {
last = error;
log.warn("fleet health: failTarget attempt {}/{} failed for member={} state={}",
attempt, MAX_FAIL_TARGET_ATTEMPTS, target, state, error);
}
}
log.warn("fleet health: giving up on failTarget for member={} state={} after {} attempts",
target, state, MAX_FAIL_TARGET_ATTEMPTS, last);
}
private static boolean terminal(HealthState state) {
return state == HealthState.GONE || state == HealthState.NEVER_READY;
}
private static boolean fault(HealthState state) {
return switch (state) {
case NEVER_READY, GONE, TURN_BOUNDARY_LOST, ERROR_ON_SCREEN, STALL_SUSPECTED,
MUTE, REPLY_STRANDED, DELEGATION_ORPHANED, CONTROL_LINK_DOWN -> true;
default -> false;
};
}
public static String coverage(boolean enabled, boolean notificationConfigured) {
return !enabled ? "off" : notificationConfigured ? "full" : "detection-only";
}
}
@@ -1,4 +0,0 @@
package dev.ltms.bridged.health;
/** Classification plus the private fact that the next pure decision needs. */
public record HealthDecision(HealthState state, HealthPrior prior) { }
@@ -1,6 +0,0 @@
package dev.ltms.bridged.health;
/** Private cross-tick observation. It is deliberately not a reported health value. */
public record HealthPrior(boolean busyButDone) {
public static final HealthPrior NONE = new HealthPrior(false);
}
@@ -1,11 +0,0 @@
package dev.ltms.bridged.health;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.session.MemberSession;
/** Read-only facts from one fleet collection tick. */
public record HealthSnapshot(MemberSession.State sessionState, AgentStatus liveStatus,
boolean acceptedDelivery, boolean queuedDelivery, boolean inboxMessage,
boolean present, boolean targetNotFound, boolean controlLinkDown,
boolean readinessGraceElapsed, boolean orphanedDelegation,
boolean replyStranded, boolean stalled) { }
@@ -1,8 +0,0 @@
package dev.ltms.bridged.health;
/** Health classifications reported for a member. */
public enum HealthState {
STARTING, IDLE, WORKING, WORK_PENDING, BLOCKED_AMBIGUOUS,
NEVER_READY, GONE, TURN_BOUNDARY_LOST, ERROR_ON_SCREEN, STALL_SUSPECTED,
MUTE, REPLY_STRANDED, DELEGATION_ORPHANED, CONTROL_LINK_DOWN
}
@@ -1,25 +0,0 @@
package dev.ltms.bridged.health;
import dev.ltms.bridged.msg.Rendezvous;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Counts turns that ended via the completion fallback instead of {@code bridge_reply}.
* MUTE is an observation by target and profile, not a classifier state and never suppresses faults.
*/
public final class MuteCounter {
private final Map<String, Integer> byTarget = new ConcurrentHashMap<>();
private final Map<String, Integer> byProfile = new ConcurrentHashMap<>();
/** Record only fallback completion; a structured reply does not make a member mute. */
public void observe(String target, String profile, Rendezvous.Kind kind) {
if (kind != Rendezvous.Kind.COMPLETION) return;
byTarget.merge(target, 1, Integer::sum);
byProfile.merge(profile, 1, Integer::sum);
}
public int forTarget(String target) { return byTarget.getOrDefault(target, 0); }
public int forProfile(String profile) { return byProfile.getOrDefault(profile, 0); }
}
@@ -1,26 +0,0 @@
package dev.ltms.bridged.health;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/** Fixed pane-probe limits. Pane content is never retained here. */
public final class PaneBudget {
public static final long COOLDOWN_NANOS = 60_000_000_000L;
public static final int MAX_PER_TICK = 2;
private final Map<String, Long> lastProbe = new HashMap<>();
private int cursor;
public List<String> choose(List<String> candidates, long nowNanos, long configuredCooldownNanos) {
long cooldown = Math.max(COOLDOWN_NANOS, configuredCooldownNanos);
List<String> out = new ArrayList<>();
for (int n = 0; n < candidates.size() && out.size() < MAX_PER_TICK; n++) {
String target = candidates.get((cursor + n) % candidates.size());
Long last = lastProbe.get(target);
if (last == null || nowNanos - last >= cooldown) { out.add(target); lastProbe.put(target, nowNanos); }
}
if (!candidates.isEmpty()) cursor = (cursor + 1) % candidates.size();
return List.copyOf(out);
}
}
@@ -6,7 +6,6 @@ import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.function.LongSupplier;
@@ -24,15 +23,6 @@ import java.util.function.Supplier;
* by first starting the session and asking it. Scanning closes that loop: label the tab, and the
* pane is recognised on the next resolve.
*
* <p><strong>CB-579 — matched by name, not prefix.</strong> This used to strip one shared
* {@code tabPrefix} off a label to derive the lead's name, and merged a config-supplied
* {@code terminal_id} pin over every scan result so the pin could never expire. Both are gone: each
* lead now configures its own exact {@code tab} label ({@code fleet.leaders.<name>.tab}), so this
* class is handed a {@code tab → name} map up front and matches labels against it exactly
* (case-insensitively). There is no merge step — a scan result is the whole answer. That is the
* fix for the bug this replaces: a {@code terminal_id} pin surviving in config after the pane it
* named was gone, so the daemon kept treating a dead session as a live lead forever.
*
* <p><strong>Direction of trust.</strong> The label names the lead; it never <em>grants</em>
* anything a pane could take for itself. Three properties keep that honest:
* <ol>
@@ -58,7 +48,7 @@ import java.util.function.Supplier;
* ever make a decision that <em>removes</em> something based on this map, add the same check.
* The remaining hazard is an <em>operator</em> one — a worker {@code tabLabel} template that
* happens to start with the same prefix would promote the whole fleet — and that is refused at
* startup by {@code BridgedConfig.validateLeadTabPrefixes} rather than documented here.
* startup by {@code BridgedConfig.validateLeadScan} rather than documented here.
*
* <p><strong>Caching.</strong> {@link #get()} is on the request path (every resolve), so the scan
* is TTL-cached and a stale-but-valid map is preferred to a herdr round-trip. A failed scan keeps
@@ -70,46 +60,38 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
private static final Logger log = LoggerFactory.getLogger(LeadTabScanner.class);
private final HerdrClient herdr;
private final Map<String, String> tabToName;
private final String tabPrefix;
private final Set<String> excludedWorkspaceLabels;
private final Map<String, String> configuredLeads;
private final long ttlNanos;
private final LongSupplier clock;
private Map<String, String> cached = Map.of();
private Map<String, String> cached;
private long scannedAtNanos;
private boolean everScanned;
/**
* @param herdr the herdr client to query ({@code workspace.list},
* {@code tab.list}, {@code pane.list} — all read-only)
* @param tabToName every configured lead's exact tab label → its name
* ({@code fleet.leaders.<name>.tab}), matched case-insensitively
* @param tabPrefix a tab whose label starts with this (case-insensitively) hosts a
* lead; the rest of the label, trimmed, is the lead's name
* @param excludedWorkspaceLabels workspaces never scanned — the configured worker spaces
* @param configuredLeads the static {@code leaders:}/{@code primary:} registry, merged
* over every scan result. Explicit config outranks the
* convention, and survives a scan that cannot run at all
* @param ttlNanos how long a scan result is reused before the next one
* @param clock nanosecond time source ({@code System::nanoTime} in production)
*/
public LeadTabScanner(HerdrClient herdr, Map<String, String> tabToName,
Set<String> excludedWorkspaceLabels, long ttlNanos, LongSupplier clock) {
public LeadTabScanner(HerdrClient herdr, String tabPrefix, Set<String> excludedWorkspaceLabels,
Map<String, String> configuredLeads, long ttlNanos, LongSupplier clock) {
this.herdr = herdr;
this.tabToName = normalize(tabToName);
this.tabPrefix = tabPrefix == null || tabPrefix.isBlank() ? "lead:" : tabPrefix.strip();
this.excludedWorkspaceLabels = excludedWorkspaceLabels == null
? Set.of() : Set.copyOf(excludedWorkspaceLabels);
this.configuredLeads = configuredLeads == null ? Map.of() : Map.copyOf(configuredLeads);
this.ttlNanos = ttlNanos;
this.clock = clock;
}
/** Keys stripped and lower-cased once, so every lookup is a plain map hit. */
private static Map<String, String> normalize(Map<String, String> tabToName) {
if (tabToName == null || tabToName.isEmpty()) {
return Map.of();
}
Map<String, String> out = new LinkedHashMap<>();
tabToName.forEach((tab, name) -> {
if (tab != null && !tab.isBlank() && name != null && !name.isBlank()) {
out.put(tab.strip().toLowerCase(Locale.ROOT), name);
}
});
return Collections.unmodifiableMap(out);
this.cached = this.configuredLeads;
}
/**
@@ -169,20 +151,25 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
}
}
}
byTerminal.putAll(configuredLeads); // an explicit pin outranks a label
return Collections.unmodifiableMap(byTerminal);
}
/**
* The lead name a tab label declares, or {@code null} if it names none of the configured leads.
* The lead name a tab label declares, or {@code null} if it declares none.
*
* <p>Exact match (case-insensitive, ends stripped) against {@link #tabToName} — no prefix
* stripping, so an operator's {@code "lead: something-else"} tab is never mistaken for a
* configured lead just because it shares a prefix.
* <p>{@code "lead: opus-5.0"} → {@code "opus-5.0"}. A bare {@code "lead:"} names nobody and is
* rejected: an unnamed lead would resolve as {@code PRIMARY} with nothing to attribute it to.
*/
private String leadNameOf(String label) {
if (label == null) {
return null;
}
return tabToName.get(label.strip().toLowerCase(Locale.ROOT));
String l = label.strip();
if (!l.regionMatches(true, 0, tabPrefix, 0, tabPrefix.length())) {
return null;
}
String name = l.substring(tabPrefix.length()).strip();
return name.isEmpty() ? null : name;
}
}
@@ -2,17 +2,11 @@ package dev.ltms.bridged.inject;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.msg.Rendezvous;
import dev.ltms.bridged.msg.TurnToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
/**
* The CB-106 completion fallback: bridges the {@link Injector}'s turn-completion signal to the
@@ -66,7 +60,6 @@ public final class CompletionResolver implements TurnListener {
private final AgentControl agents;
private final Rendezvous rendezvous;
private final ExhaustedPatternLookup exhaustedPatterns;
/**
* Per-target record of the turn currently in flight: the exact {@link Rendezvous} waiter its
@@ -84,30 +77,23 @@ public final class CompletionResolver implements TurnListener {
private final ConcurrentHashMap<String, InFlight> inFlight = new ConcurrentHashMap<>();
/**
* @param exhaustedPatterns CB-578 stage A: per-target lookup for a profile's configured
* 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()}).
*/
public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns) {
public CompletionResolver(AgentControl agents, Rendezvous rendezvous) {
this.agents = agents;
this.rendezvous = rendezvous;
this.exhaustedPatterns = Objects.requireNonNull(exhaustedPatterns, "exhaustedPatterns");
}
@Override
public void onDelivered(String target, TurnToken token) {
public void onDelivered(String target) {
// Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn
// content — what it shows *before* the just-delivered turn produces output — as the staleness
// reference (CB-115). Done synchronously (like the delivering send itself) so both are in
// place before this turn's completion can fire.
captureBaseline(target, token);
captureBaseline(target);
}
/** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
void captureBaseline(String target, TurnToken token) {
CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
void captureBaseline(String target) {
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
if (waiter == null) {
inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later
return;
@@ -151,13 +137,7 @@ public final class CompletionResolver implements TurnListener {
@Override
public void onTurnFailed(String target) {
InFlight turn = inFlight.get(target);
Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn, null));
}
@Override
public void onTurnFailed(String target, String reason) {
InFlight turn = inFlight.get(target);
Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn, reason));
Thread.ofVirtual().name("turn-failed-" + target).start(() -> fail(target, turn));
}
/** Synchronous resolve (the unit-testable core of {@link #onTurnComplete}). */
@@ -170,12 +150,11 @@ public final class CompletionResolver implements TurnListener {
return;
}
String tail;
String assistantBlock = null;
int originalLength = 0;
boolean clipped = false;
boolean scrapeFailed = false;
try {
assistantBlock = lastAssistantBlock(agents.read(target, SCRAPE_SOURCE));
String assistantBlock = lastAssistantBlock(agents.read(target, SCRAPE_SOURCE));
originalLength = assistantBlock.strip().length();
clipped = originalLength > MAX_SCRAPE_CHARS;
tail = clip(assistantBlock);
@@ -198,22 +177,6 @@ public final class CompletionResolver implements TurnListener {
target);
return; // keep the in-flight record: a later genuine completion still needs it
}
// CB-578 stage A: a turn that ended with no bridge_reply AND whose scrape matches the
// backend's configured usage-limit pattern is a refusal, not an answer. Classify it as
// BACKEND_EXHAUSTED rather than handing the caller a scrape that reads like a real reply.
if (!scrapeFailed) {
Pattern exhausted = exhaustedPatterns.patternFor(target);
String matchedLine = exhausted == null ? null : firstMatchingLine(assistantBlock, exhausted);
if (matchedLine != null) {
String reason = "backend exhausted (usage limit): " + matchedLine;
if (rendezvous.resolveExhausted(waiter, reason)) {
inFlight.remove(target, turn);
log.warn("completion for {} classified BACKEND_EXHAUSTED (no bridge_reply; scrape "
+ "matched the profile's exhausted pattern): {}", target, reason);
}
return;
}
}
String completion = clipped ? tail + "\n" + CLIPPED_PANE_TAIL_MARKER : tail;
if (rendezvous.resolveCompletion(waiter, completion)) {
inFlight.remove(target, turn);
@@ -229,11 +192,6 @@ public final class CompletionResolver implements TurnListener {
/** Synchronous fail (the unit-testable core of {@link #onTurnFailed}). */
void fail(String target, InFlight turn) {
fail(target, turn, null);
}
/** Synchronous fail with an optional reason supplied by a dropped worker queue. */
void fail(String target, InFlight turn, String explicitReason) {
// A never-delivered readiness failure has no in-flight record but still has a blocked send;
// fall back to the currently-registered waiter (unambiguous — that send never completed, so
// no next turn exists to confuse it with).
@@ -243,18 +201,16 @@ public final class CompletionResolver implements TurnListener {
inFlight.remove(target, turn); // nobody blocked on this worker — nothing to fail
return;
}
String reason = explicitReason;
if (reason == null || reason.isBlank()) {
try {
reason = clip(agents.read(target, SCRAPE_SOURCE));
} catch (RuntimeException e) {
reason = "";
}
if (reason.isBlank()) {
// No screen to scrape — either the worker is stuck (CB-109) or gone (CB-110).
reason = "worker did not reply; its turn ended in an unrecoverable state "
+ "(worker unreachable or stuck)";
}
String reason;
try {
reason = clip(agents.read(target, SCRAPE_SOURCE));
} catch (RuntimeException e) {
reason = "";
}
if (reason.isBlank()) {
// No screen to scrape — either the worker is stuck (CB-109) or gone (CB-110).
reason = "worker did not reply; its turn ended in an unrecoverable state "
+ "(worker unreachable or stuck)";
}
if (rendezvous.resolveFailure(waiter, reason)) {
inFlight.remove(target, turn);
@@ -262,45 +218,6 @@ public final class CompletionResolver implements TurnListener {
}
}
/**
* The first line of {@code text} matching {@code pattern}, stripped — the CB-578 stage A
* evidence carried in a {@code BACKEND_EXHAUSTED} reason so the operator sees the real refusal
* text, never a generic label. {@code null} if no line matches.
*/
static String firstMatchingLine(String text, Pattern pattern) {
if (text == null || text.isEmpty()) return null;
for (String line : text.split("\n", -1)) {
if (pattern.matcher(line).find()) {
return line.strip();
}
}
return null;
}
/**
* Coverage summary for the CB-578 stage A exhausted-pattern classification, logged at startup
* the way {@link dev.ltms.bridged.health.FleetHealthMonitor#coverage} is — so an operator can
* see whether the classification is on, and for which profiles, without reading every
* profile's config by hand.
*
* @param allProfiles every configured profile name
* @param configuredProfiles the subset of {@code allProfiles} that carry an exhausted pattern
*/
public static String coverage(Set<String> allProfiles, Set<String> configuredProfiles) {
if (configuredProfiles.isEmpty()) {
return "off (no profile has an exhaustedPattern configured; profiles: " + sorted(allProfiles) + ")";
}
Set<String> unconfigured = new TreeSet<>(allProfiles);
unconfigured.removeAll(configuredProfiles);
return unconfigured.isEmpty()
? "full (all profiles configured: " + sorted(allProfiles) + ")"
: "partial (configured: " + sorted(configuredProfiles) + "; not configured: " + sorted(unconfigured) + ")";
}
private static List<String> sorted(Set<String> names) {
return names.stream().sorted().toList();
}
private static String clip(String s) {
if (s == null) return "";
String trimmed = s.strip();
@@ -1,27 +0,0 @@
package dev.ltms.bridged.inject;
import java.util.regex.Pattern;
/**
* Per-target lookup for a profile's configured usage-limit refusal pattern (CB-578 stage A): how
* {@link CompletionResolver} tells a backend that refused on a subscription usage limit — the
* worker's pane stays healthy, but the account is exhausted — apart from a genuine completion.
*
* <p>The pattern is always profile config, never a vendor string in Java source: every backend
* words its refusal differently, so a hardcoded sentence would only ever match one of them.
*/
@FunctionalInterface
public interface ExhaustedPatternLookup {
/** The compiled pattern configured for {@code target}'s profile, or {@code null} if none. */
Pattern patternFor(String target);
/**
* Inert lookup — no profile has a pattern configured, so the classification never fires and
* the completion fallback behaves exactly as before CB-578 stage A. The explicit stand-in a
* caller (or a test not exercising this feature) passes instead of a defaulting overload.
*/
static ExhaustedPatternLookup none() {
return target -> null;
}
}
@@ -2,7 +2,6 @@ package dev.ltms.bridged.inject;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.msg.TurnToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -130,7 +129,7 @@ public final class Injector {
}
/** A pending message and the future that completes when it has been delivered. */
private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
private record Pending(String text, CompletableFuture<Void> delivered) {
}
/** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -160,9 +159,9 @@ public final class Injector {
* <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
* target" and "queue the message" and orphan it in a target it just removed.
*/
public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
public CompletableFuture<Void> enqueue(String target, String text) {
CompletableFuture<Void> delivered = new CompletableFuture<>();
Pending p = new Pending(text, token, delivered);
Pending p = new Pending(text, delivered);
targets.compute(target, (_, existing) -> {
Target t = (existing != null) ? existing : new Target();
t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
@@ -357,7 +356,7 @@ public final class Injector {
} else {
// Baseline the pane's pre-turn content so a misattributed completion (no new output)
// can't resolve this send with the previous turn's stale answer (CB-115).
turnListener.onDelivered(target, sent.token());
turnListener.onDelivered(target);
sent.delivered().complete(null);
}
}
@@ -411,8 +410,8 @@ public final class Injector {
for (Pending p : pending) {
p.delivered().completeExceptionally(cause);
}
// A queued send has no in-flight record, while a delivered turn does. CompletionResolver
// handles both forms and resolves its waiter at most once.
turnListener.onTurnFailed(target, cause.getMessage());
if (hadDeliveredTurn) {
turnListener.onTurnFailed(target);
}
}
}
@@ -1,7 +1,5 @@
package dev.ltms.bridged.inject;
import dev.ltms.bridged.msg.TurnToken;
/**
* Notified when a worker's delegated turn is observed to complete — a confirmed
* {@code WORKING → IDLE} transition after a delivery. This is the CB-106 completion signal the
@@ -43,14 +41,6 @@ public interface TurnListener {
default void onTurnFailed(String target) {
}
/**
* As {@link #onTurnFailed(String)}, carrying the reason a worker became unreachable. The default
* keeps existing listeners working while allowing the completion resolver to report a useful cause.
*/
default void onTurnFailed(String target, String reason) {
onTurnFailed(target);
}
/**
* A message was just delivered into {@code target}'s pane (CB-115). Fired so the completion
* resolver can snapshot the pane's pre-turn content: a later {@link #onTurnComplete} whose
@@ -59,7 +49,7 @@ public interface TurnListener {
* resolve the send with the previous turn's stale answer. A default no-op keeps the interface
* functional for callers that don't scrape.
*/
default void onDelivered(String target, TurnToken token) {
default void onDelivered(String target) {
}
/** No-op default for callers that only need delivery, not completion signalling. */
@@ -59,7 +59,7 @@ public final class LeadLauncher {
/**
* @param agents herdr agent control (start, list)
* @param spaces workspace / tab control (ensure, create, label, list)
* @param cfg the loaded config — {@code fleet.leaders}, {@code profiles} and each lead's tab
* @param cfg the loaded config — {@code fleet.leaders}, {@code profiles} and the lead pins
*/
public LeadLauncher(AgentControl agents, WorkspaceControl spaces, BridgedConfig cfg) {
this.agents = agents;
@@ -102,8 +102,8 @@ public final class LeadLauncher {
continue;
}
if (!lead.isCreatable()) {
// A lead with a `tab:` but no `profile:` is recognise-only by design: the operator
// opens it by hand. Say so once rather than looking like a silent failure.
// A lead with a `terminal:` pin and no `profile:` is recognise-only by design: the
// operator opens it by hand. Say so once rather than looking like a silent failure.
log.info("lead '{}' is not live, and names no profile — it can be recognised but not "
+ "launched. Add `profile:` under fleet.leaders.{} to have bridged start it.",
name, name);
@@ -127,16 +127,18 @@ public final class LeadLauncher {
}
/**
* How many live leads exist per configured name: a running agent in a tab labelled with that
* lead's exact {@code tab} (CB-579). Member workspaces are excluded, exactly as the scanner
* excludes them: a member must not be counted as a lead because it happens to sit in a matching
* tab.
* How many live leads exist per configured name.
*
* <p>There used to be a second path here — a running agent on the terminal a
* {@code fleet.leaders.<name>.terminal} pin named, for a lead opened and pinned by hand. That
* pin is retired: {@code tab} is now the only field identity depends on, and {@link Agent}
* already carries {@link Agent#tabId()} directly, so a hand-opened lead is found the same way an
* auto-launched one is — by labelling its tab to match.
* <p>Two independent pieces of evidence, because either alone double-spawns:
* <ul>
* <li>a running agent in a tab labelled {@code "<tabPrefix> <name>"} — how an auto-launched
* lead, or an operator following the labelling convention, is found;</li>
* <li>a running agent on a terminal the config pins in {@code fleet.leaders.<name>.terminal} —
* how a lead the operator opened and pinned by hand is found. Without this, a pinned lead
* whose tab carries no matching label would be relaunched on every boot.</li>
* </ul>
* Member workspaces are excluded, exactly as the scanner excludes them: a member must not be
* counted as a lead because it happens to sit in a matching tab.
*/
private Map<String, Integer> liveLeads(Map<String, BridgedConfig.Leader> leaders) {
Set<String> memberSpaces = cfg.profiles().values().stream()
@@ -158,9 +160,20 @@ public final class LeadLauncher {
}
}
// terminalId → the lead name the config pins it to.
Map<String, String> nameByPinnedTerminal = new LinkedHashMap<>();
leaders.forEach((name, lead) -> {
if (lead.terminal() != null && !lead.terminal().isBlank()) {
nameByPinnedTerminal.put(lead.terminal().strip(), name);
}
});
Map<String, Integer> counts = new LinkedHashMap<>();
for (Agent a : agents.list()) {
String name = nameByTab.get(a.tabId());
if (name == null) {
name = nameByPinnedTerminal.get(a.terminalId());
}
if (name != null) {
counts.merge(name, 1, Integer::sum);
}
@@ -171,7 +184,7 @@ public final class LeadLauncher {
/**
* The configured lead a tab label names, or {@code null} for a label that names none.
*
* <p>Matched exactly (case-insensitively) against each lead's configured {@code tab}, so an
* <p>Matched against the declared lead names rather than by splitting on the prefix, so an
* operator's {@code "lead: something-else"} tab is not mistaken for a configured lead.
*/
private String leadNameOf(String label, Map<String, BridgedConfig.Leader> leaders) {
@@ -180,8 +193,7 @@ public final class LeadLauncher {
}
String l = label.strip();
for (Map.Entry<String, BridgedConfig.Leader> e : leaders.entrySet()) {
String tab = e.getValue().tabLabel();
if (tab != null && l.equalsIgnoreCase(tab.strip())) {
if (l.equalsIgnoreCase(e.getValue().tabLabel(e.getKey()).strip())) {
return e.getKey();
}
}
@@ -190,7 +202,7 @@ public final class LeadLauncher {
/** Start one lead. Returns false (having logged) rather than throwing on any failure. */
private boolean launch(String name, BridgedConfig.Leader lead, BridgedConfig.Profile profile) {
String label = lead.tabLabel();
String label = lead.tabLabel(name);
String cwd = (lead.cwd() == null || lead.cwd().isBlank())
? System.getProperty("user.dir") : lead.cwd();
@@ -1,30 +0,0 @@
package dev.ltms.bridged.logging;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.turbo.TurboFilter;
import ch.qos.logback.core.spi.FilterReply;
import io.modelcontextprotocol.spec.McpSchema;
import org.slf4j.Marker;
/** Suppresses only the SDK warning for the normal MCP cancellation notification. */
public final class McpCancelledNotificationFilter extends TurboFilter {
static final String LOGGER = "io.modelcontextprotocol.spec.McpStreamableServerSession";
static final String UNHANDLED_NOTIFICATION = "No handler registered for notification method: {}";
@Override
public FilterReply decide(Marker marker, Logger logger, Level level, String format, Object[] params,
Throwable throwable) {
if (level == Level.WARN
&& LOGGER.equals(logger.getName())
&& UNHANDLED_NOTIFICATION.equals(format)
&& params != null
&& params.length == 1
&& params[0] instanceof McpSchema.JSONRPCNotification notification
&& "notifications/cancelled".equals(notification.method())) {
return FilterReply.DENY;
}
return FilterReply.NEUTRAL;
}
}
@@ -33,10 +33,7 @@ import jakarta.servlet.http.HttpServlet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
import java.util.stream.Collectors;
/**
@@ -76,20 +73,17 @@ public final class BridgeMcp {
private final McpSyncServer server;
private final CallerResolver authz; // CB-501: null → authorization not enforced (legacy)
private final Metrics metrics; // CB-502: null → auth failures not counted
private final CapacitySource capacity;
private final HealthCoverageSource healthCoverage;
/** Capacity facts used by {@code bridge_list}; production must supply the placement live count. */
public record CapacitySource(Function<String, Integer> liveCount, Function<String, Integer> maxLoad,
Supplier<Set<String>> configuredProfiles, LongSupplier clock) {
/** Inert test-only source. It omits capacity rather than inventing zero live counts. */
public static CapacitySource none() { return new CapacitySource(_ -> 0, _ -> null, Set::of, System::nanoTime); }
boolean available() { return !configuredProfiles.get().isEmpty(); }
/**
* Legacy constructor — no authorization. Retained so existing tests exercise tool behaviour
* without an auth fixture.
*/
public BridgeMcp(MessageService messages, PeerLauncher workers,
SessionManager sessions, ConnectionIdentity identity, MemberPresence presence,
PrimaryRegistry primaryRegistry) {
this(messages, workers, sessions, identity, presence, primaryRegistry, null, null);
}
/** Coverage is supplied by the health wiring, not inferred from a missing dependency. */
public record HealthCoverageSource(Supplier<String> value) { }
/**
* @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
@@ -97,11 +91,9 @@ public final class BridgeMcp {
* filter, so the REST guard does not cover it.
* @param metrics registry for auth-failure counting; may be {@code null}
*/
public BridgeMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage) {
this.capacity = capacity;
this.healthCoverage = healthCoverage;
public BridgeMcp(MessageService messages, PeerLauncher workers,
SessionManager sessions, ConnectionIdentity identity, MemberPresence presence,
PrimaryRegistry primaryRegistry, CallerResolver callers, Metrics metrics) {
McpJsonMapper json = new JacksonMcpJsonMapperSupplier().get();
this.transport = HttpServletStreamableServerTransportProvider.builder()
.jsonMapper(json)
@@ -159,8 +151,8 @@ public final class BridgeMcp {
Runnable onAccepted = () -> primaryRegistry.recordDelegation(target, caller);
// wait defaults to true (block for the reply); wait:false is fire-and-poll.
return Boolean.FALSE.equals(a.get("wait"))
? sendAsync(messages, target, content, onAccepted, workers.profiles())
: send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles());
? sendAsync(messages, target, content, onAccepted)
: send(messages, target, content, timeoutMs(a), onAccepted);
})
// bridge_reply's identity is the CONNECTION, never an argument — so the authz check
// is "is this caller a worker at all", and it can only ever reply as itself.
@@ -215,7 +207,7 @@ public final class BridgeMcp {
.toolCall(listTool(), (exchange, _) -> {
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
if (denied != null) return denied;
return listFleet(workers, sessions, messages, capacity, healthCoverage,
return listFleet(workers, sessions,
callers == null ? Map.of() : callers.leads(),
callerTerminal(exchange));
})
@@ -387,22 +379,21 @@ public final class BridgeMcp {
// --- tool logic (thin adapters over the services; unit-testable) ---------------------------
/** {@code bridge_send}: delegate {@code content} to a worker session and block for its reply. */
static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content, Long timeoutMs) {
return send(messages, sessionId, content, timeoutMs, null);
}
/**
* {@code bridge_send}: delegate {@code content} to a worker session and block for its reply.
* The configured profiles are required so a profile name can never bypass target validation.
*
* As {@link #send(MessageService, String, String, Long)}, wiring an accepted-delivery hook
* (CB-548): {@code onAccepted} records delegator ownership the instant the send is accepted, so
* a BUSY interloper never claims a turn it did not win. {@code null} disables recording.
*/
static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content,
Long timeoutMs, Runnable onAccepted, Set<String> profiles) {
Long timeoutMs, Runnable onAccepted) {
if (isBlank(sessionId) || isBlank(content)) {
return error("sessionId and content are required");
}
McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
if (targetError != null) {
return targetError;
}
long timeout = clamp(timeoutMs == null ? DEFAULT_TIMEOUT_MS : timeoutMs);
try {
return formatReply(messages.send(sessionId, content, timeout, onAccepted), timeout);
@@ -458,11 +449,6 @@ public final class BridgeMcp {
"[worker finished without a structured bridge_reply — transcript tail follows]\n" + r.text());
// The worker ran the turn then wedged (CB-109) — surface the error context.
case WORKER_FAILED -> text("[worker failed — turn ended in an unrecoverable state]\n" + r.text());
// The backend refused on a subscription usage limit (CB-578 stage A) — the worker's
// pane stayed healthy, but its account is exhausted. Distinct from WORKER_FAILED so the
// primary gets the real cause, not a generic wedge.
case BACKEND_EXHAUSTED -> text("[backend exhausted — the worker's account refused on a "
+ "usage limit]\n" + r.text());
// The worker paused mid-turn to ask (CB-205) — tell the primary how to answer in-turn.
case QUESTION -> text("[question] the worker paused to ask before it can finish:\n" + r.text()
+ "\n\nAnswer it by calling bridge_send again with turnId=\"" + r.turnId()
@@ -477,33 +463,24 @@ public final class BridgeMcp {
/**
* {@code bridge_send} with {@code wait:false}: delegate {@code content} and return a ticket
* immediately (fire-and-poll), so a long task isn't cut off by the caller's MCP call timeout.
* The configured profiles are required so a profile name can never bypass target validation.
*
* This wires the accepted-delivery hook
*/
static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content) {
return sendAsync(messages, sessionId, content, null);
}
/**
* As {@link #sendAsync(MessageService, String, String)}, wiring the accepted-delivery hook
* (CB-548) so an async flooding send records delegator ownership exactly once it is accepted.
*/
static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content,
Runnable onAccepted, Set<String> profiles) {
Runnable onAccepted) {
if (isBlank(sessionId) || isBlank(content)) {
return error("sessionId and content are required");
}
McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
if (targetError != null) {
return targetError;
}
String ticket = messages.sendAsync(sessionId, content, onAccepted);
return text("accepted — task delegated. Poll bridge_poll with ticket=" + ticket);
}
/** A configured profile is never a send target; other unknown values may be herdr-owned panes. */
private static McpSchema.CallToolResult profileTargetError(String sessionId, Set<String> profiles) {
if (profiles.contains(sessionId)) {
return error("unknown send target \"" + sessionId + "\": it is a configured profile name, not a "
+ "session id. Call bridge_list to find a member or lead sessionId.");
}
return null;
}
/** {@code bridge_poll}: check an async delegation by ticket, or drain a worker's inbox by target. */
static McpSchema.CallToolResult poll(MessageService messages, String ticket, String target) {
if (!isBlank(target)) {
@@ -525,9 +502,6 @@ public final class BridgeMcp {
? "[done — worker finished without a structured bridge_reply; transcript tail follows]\n" + v.reply()
: v.reply());
case PENDING -> text("[pending — " + v.detail() + "]");
case ASKING -> text("[question — worker is waiting for your answer]\n" + v.reply()
+ "\n\nAnswer it by calling bridge_send again with turnId=\"" + v.turnId()
+ "\" and content set to your answer; the worker resumes the same turn.");
case FAILED -> text("[failed — " + v.detail() + "]");
};
}
@@ -734,12 +708,6 @@ public final class BridgeMcp {
*/
static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions,
Map<String, String> leads, String selfTerm) {
return listFleet(workers, sessions, null, CapacitySource.none(), new HealthCoverageSource(() -> "off"), leads, selfTerm);
}
static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
CapacitySource capacity, HealthCoverageSource healthCoverage,
Map<String, String> leads, String selfTerm) {
try {
Map<String, Agent> live = workers.list().stream()
.map(Agent.class::cast)
@@ -749,57 +717,15 @@ public final class BridgeMcp {
.sorted(Map.Entry.comparingByValue())
.map(e -> leadView(e.getKey(), e.getValue(), live.get(e.getKey()), selfTerm))
.toList();
List<MemberSession> roster = sessions.roster();
List<Map<String, Object>> out = roster.stream()
.map(s -> memberCapacityView(s, live.get(s.terminalId()), messages, capacity.clock().getAsLong()))
List<Map<String, Object>> out = sessions.roster().stream()
.map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
.toList();
Set<String> profiles = new java.util.TreeSet<>(capacity.configuredProfiles().get());
roster.stream().map(MemberSession::profile).forEach(profiles::add);
Map<String, Object> result = new LinkedHashMap<>();
result.put("leads", leadRows); result.put("members", out);
result.put("healthCoverage", healthCoverage.value().get());
if (capacity.available()) result.put("capacity", profiles.stream()
.map(profile -> capacityView(profile, capacity.liveCount(), capacity.maxLoad(), roster, messages,
capacity.clock().getAsLong())).toList());
return text(json(result));
return text(json(Map.of("leads", leadRows, "members", out)));
} catch (HerdrException e) {
return error("herdr error listing the fleet: " + e.getMessage());
}
}
/**
* Capacity is advisory only. {@code reclaimable} says there is no bridge work, not that bridged
* may stop the member: the bridge has capacity facts but no work list, and choosing work needs
* authority it does not have. {@code idleForSeconds} is derived from monotonic nanoTime and has
* no wall-clock meaning across a daemon restart.
*/
private static Map<String, Object> memberCapacityView(MemberSession session, Agent live,
MessageService messages, long nowNanos) {
Map<String, Object> row = SessionManager.rosterView(session, live);
boolean open = messages != null && messages.hasAcceptedDelivery(session.terminalId());
boolean inbox = messages != null && messages.hasInboxMessage(session.terminalId());
boolean reclaimable = (session.state() == MemberSession.State.READY || session.state() == MemberSession.State.DONE)
&& !open && !inbox;
row.put("reclaimable", reclaimable);
row.put("idleForSeconds", reclaimable ? Math.max(0, (nowNanos - session.lastActivityAtNanos()) / 1_000_000_000L) : null);
return row;
}
private static Map<String, Object> capacityView(String profile, Function<String, Integer> liveCount,
Function<String, Integer> maxLoad, List<MemberSession> roster,
MessageService messages, long nowNanos) {
Integer cap = maxLoad.apply(profile);
int live = liveCount.apply(profile);
int reclaimable = (int) roster.stream().filter(s -> profile.equals(s.profile()))
.filter(s -> (s.state() == MemberSession.State.READY || s.state() == MemberSession.State.DONE))
.filter(s -> messages == null || (!messages.hasAcceptedDelivery(s.terminalId()) && !messages.hasInboxMessage(s.terminalId())))
.count();
Map<String, Object> row = new LinkedHashMap<>();
row.put("profile", profile); row.put("maxLoad", cap); row.put("live", live);
row.put("free", cap == null ? null : Math.max(0, cap - live)); row.put("reclaimable", reclaimable);
return row;
}
/**
* One lead's row: its address, its name, and whether it can be reached right now.
*
@@ -44,6 +44,23 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
private final SubscriptionGuard guard;
/**
* Standing instruction appended to the worker's system prompt so it returns its result via
* {@code bridge_reply}. Injected as a launch flag, so nothing is written to the worker's
* profile — it is guidance, and a worker that never replies is caught by the send's timeout.
*/
static final String REPLY_CHARTER =
"You are an off-subscription worker in the claude-bridge fleet. Every message you "
+ "receive arrives through the bridge, and the ONLY channel back to the sender is the "
+ "bridge_reply MCP tool. Text you write in your terminal is NOT sent anywhere — the "
+ "sender cannot see your screen, so an in-terminal answer is silently discarded. "
+ "Therefore you MUST end EVERY turn by calling bridge_reply with `content` set to your "
+ "complete response. This holds for every message without exception — tasks, questions, "
+ "clarifications, acknowledgements, and ordinary back-and-forth conversation. Call "
+ "bridge_reply exactly once, as the final action of your turn, with your full answer in "
+ "`content`; never wait for confirmation first. If you end a turn without calling "
+ "bridge_reply, the sender receives nothing and the exchange stalls.";
/**
* Production constructor — disables the spawn-ready gate ({@code spawnReadyTimeoutMs == 0}) so
* existing deployments and tests keep the legacy non-blocking spawn semantics.
@@ -76,11 +93,11 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
Function<String, String> env,
long spawnReadyTimeoutMs, long spawnReadyPollMs,
Supplier<BridgedConfig.Fleet> fleet) {
Supplier<String> tabLabelTemplate) {
this(agents, spaces, guard, profiles, defaultProfile, env,
spawnReadyTimeoutMs,
System::currentTimeMillis, () -> sleepUninterruptibly(spawnReadyPollMs),
fleet);
tabLabelTemplate);
}
/**
@@ -112,19 +129,31 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
/**
* Full testability constructor, plus the fleet-wide tab-label template (CB-557).
*
* @param fleet live fleet config, read once for each spawn
* @param tabLabelTemplate {@code fleet.tabLabel}; {@code null}/blank ⇒
* {@link BridgedConfig.Fleet#DEFAULT_TAB_LABEL}
*/
public ClaudeCodeLauncher(AgentControl agents, WorkspaceControl spaces, SubscriptionGuard guard,
Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
Function<String, String> env,
long spawnReadyTimeoutMs,
LongSupplier nowMillis, Runnable sleeper,
Supplier<BridgedConfig.Fleet> fleet) {
Supplier<String> tabLabelTemplate) {
super(NAME_PREFIX, agents, spaces, profiles, defaultProfile, env,
spawnReadyTimeoutMs, nowMillis, sleeper, fleet);
spawnReadyTimeoutMs, nowMillis, sleeper, tabLabelTemplate);
this.guard = guard;
}
/**
* {@inheritDoc}
*
* <p>A legacy spawn with no session identity is a fresh, launcher-derived session — delegate to
* the session-aware form with no name and no resume id.
*/
@Override
protected Launch buildLaunch(BridgedConfig.Profile cfg) {
return buildLaunch(cfg, null, null);
}
/**
* {@inheritDoc}
*
@@ -135,7 +164,7 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
* applied here — see {@link #applySessionIdentity}.
*/
@Override
protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {
protected Launch buildLaunch(BridgedConfig.Profile cfg, String sessionName, String resumeSessionId) {
// CB-539: a profile may deliberately opt into the subscription (subscription: true) when no
// off-subscription endpoint exists for it — e.g. `sonnet` on `ccs`. That profile gets no
// ANTHROPIC_BASE_URL/AUTH_TOKEN (there is nothing to point them at) and the guard's base_url
@@ -181,9 +210,9 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
// id via -r and passes no --session-id (the two conflict). Both are injected before the
// model flag so --model keeps outranking the operator's own argv.
// mutableArgv: argvWithBridge may hand back the profile's own (immutable) List.of when it
// has neither MCP nor a charter — session flags must be added into a list we own.
List<String> argv = mutableArgv(argvWithBridge(cfg, spec.charter()));
String agentSessionId = applySessionIdentity(argv, spec.sessionName(), spec.resumeSessionId());
// has no MCP — session flags must be added into a list we own.
List<String> argv = mutableArgv(argvWithBridge(cfg));
String agentSessionId = applySessionIdentity(argv, sessionName, resumeSessionId);
return new Launch(workerEnv, argvWithModel(argv, cfg), agentSessionId);
}
@@ -217,26 +246,22 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
}
/**
* The launch argv, plus an inline {@code --mcp-config} when {@code worker.mcpUrl} is set and
* {@code --append-system-prompt} when the base composed a charter. Neither touches the profile's
* config; both are pure command-line flags. This inline-flag mount is Claude Code specific —
* other adapters mount MCP and instructions their own way.
* The launch argv, plus — when {@code worker.mcpUrl} is set — inline {@code --mcp-config} for
* the bridge server and {@code --append-system-prompt} for the {@link #REPLY_CHARTER}. Neither
* touches the profile's config; both are pure command-line flags. This inline-flag mount is
* Claude Code specific — other adapters mount MCP and instructions their own way.
*/
private List<String> argvWithBridge(BridgedConfig.Profile cfg, String charter) {
if (!cfg.hasMcp() && charter == null) {
private List<String> argvWithBridge(BridgedConfig.Profile cfg) {
if (!cfg.hasMcp()) {
return cfg.argv();
}
String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
+ cfg.mcpUrl() + "\"}}}";
List<String> argv = mutableArgv(cfg.argv());
if (cfg.hasMcp()) {
String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
+ cfg.mcpUrl() + "\"}}}";
argv.add("--mcp-config");
argv.add(mcpJson);
}
if (charter != null) {
argv.add("--append-system-prompt");
argv.add(charter);
}
argv.add("--mcp-config");
argv.add(mcpJson);
argv.add("--append-system-prompt");
argv.add(REPLY_CHARTER);
return argv;
}
@@ -44,7 +44,7 @@ import java.util.regex.Pattern;
* <li>{@code namePrefix} (constructor arg) — the label prefix ({@code claude}, {@code opencode})
* that drives both unique naming and the orphan-reap pattern, so each adapter reaps only its
* own kind of pane and never another's.</li>
* <li>{@link #buildLaunch(BridgedConfig.Profile, LaunchSpec)} — the peer-specific env map + argv, including any
* <li>{@link #buildLaunch(BridgedConfig.Profile)} — the peer-specific env map + argv, including any
* subscription/guard check, MCP mount, and instruction injection. The base never sees how the
* peer is configured; it only places and starts the returned {@link Launch}.</li>
* </ul>
@@ -80,25 +80,15 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
private final AtomicLong nameSeq = new AtomicLong(); // per-peer counter (herdr agent names only)
/**
* Live fleet config, read once per spawn. A null supplier or value leaves tab labels at their
* default and supplies no role charter. A profile's own {@code tabLabel} still overrides it.
* The {@code fleet.tabLabel} template; a {@code null} supplier or a {@code null}/blank value ⇒
* {@link BridgedConfig.Fleet#DEFAULT_TAB_LABEL}. A profile's own {@code tabLabel} still
* overrides it.
*
* <p>CB-559: a supplier rather than a snapshot, so a config reload affects the next launch
* without a restart. Existing tabs keep the label they were given.
* <p>CB-559: a supplier rather than a String, so a config reload renames the <em>next</em> tab
* without a restart. Existing tabs keep the label they were given — bridged does not rewrite a
* label it already wrote.
*/
private final Supplier<BridgedConfig.Fleet> fleet;
/** The final instruction always requires a bridge reply when the bridge MCP is mounted. */
protected static final String REPLY_CHARTER =
"You are a spawned member in the claude-bridge fleet. Every message you receive arrives "
+ "through the bridge, and the ONLY channel back to the sender is the bridge_reply MCP tool. "
+ "Text you write in your terminal is NOT sent anywhere — the sender cannot see your screen, "
+ "so an in-terminal answer is silently discarded. Therefore you MUST end EVERY turn by calling "
+ "bridge_reply with `content` set to your complete response. This holds for every message without "
+ "exception — tasks, questions, clarifications, acknowledgements, and ordinary back-and-forth "
+ "conversation. Call bridge_reply exactly once, as the final action of your turn, with your full "
+ "answer in `content`; never wait for confirmation first. If you end a turn without calling "
+ "bridge_reply, the sender receives nothing and the exchange stalls.";
private final Supplier<String> tabLabelTemplate;
/**
* Tab numbers, counted per {@code role/profile} pair (CB-557).
@@ -152,19 +142,21 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
}
/**
* As above, plus the live {@code fleet} config (CB-557).
* As above, plus the {@code fleet.tabLabel} template (CB-557).
*
* @param fleet live fleet config, read once per spawn; {@code null} ⇒ default tab label and no
* role charter. A separate constructor rather than a new parameter on the one
* above, so every existing call site keeps the default without an edit.
* @param tabLabelTemplate fleet-wide tab-label template, read per spawn (CB-559); {@code null},
* or a supplier yielding {@code null}/blank ⇒
* {@link BridgedConfig.Fleet#DEFAULT_TAB_LABEL}. A separate constructor
* rather than a new parameter on the one above, so every existing call
* site keeps the default without an edit.
*/
protected HerdrPeerLauncher(String namePrefix, AgentControl agents, WorkspaceControl spaces,
Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
Function<String, String> env,
long spawnReadyTimeoutMs,
LongSupplier nowMillis, Runnable sleeper,
Supplier<BridgedConfig.Fleet> fleet) {
this.fleet = fleet;
Supplier<String> tabLabelTemplate) {
this.tabLabelTemplate = tabLabelTemplate;
this.namePrefix = namePrefix;
this.agents = agents;
this.spaces = spaces;
@@ -183,7 +175,22 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
* Any subscription/guard check, MCP mount, and instruction injection happen here. The env map
* and argv are adapter-private; the base only places and starts what is returned.
*/
protected abstract Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec);
protected abstract Launch buildLaunch(BridgedConfig.Profile cfg);
/**
* Session-aware variant of {@link #buildLaunch(BridgedConfig.Profile)} (CB-547a). Default
* discards the session identity and delegates to the profile-only form, so an adapter that
* carries no durable peer session (opencode, say) inherits byte-identical behaviour and needs
* no change. An adapter that does (Claude Code) overrides this to mint/resume the id and to
* surface it on the returned {@link Launch#agentSessionId()}.
*
* @param cfg the resolved profile to spawn
* @param sessionName the bridge's logical session name, or null/blank for launcher-derived
* @param resumeSessionId the peer's own prior session id to resume, or null/blank for fresh
*/
protected Launch buildLaunch(BridgedConfig.Profile cfg, String sessionName, String resumeSessionId) {
return buildLaunch(cfg);
}
/** Direct transport access for peer-specific, non-turn control operations. */
protected final AgentControl agents() {
@@ -218,10 +225,6 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
}
}
/** All per-spawn values adapters may need, including the base-composed effective charter. */
protected record LaunchSpec(String sessionName, String resumeSessionId, MemberRole role, String charter) {
}
// --- profile surface -----------------------------------------------------------------------
/** The configured peer profile names (what {@code spawn(profile)} accepts). */
@@ -292,23 +295,18 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
}
/**
* Spawn a peer with session identity (CB-547a). The session values, role, and charter are
* threaded from the {@link SpawnRequest} into {@link #buildLaunch(BridgedConfig.Profile,
* LaunchSpec)}, and the launch's resolved agent-session id is returned alongside the agent so
* the caller can put it on the {@link PeerHandle}.
* Spawn a peer with session identity (CB-547a). {@code sessionName} and {@code resumeSessionId}
* are threaded from the {@link SpawnRequest} into {@link #buildLaunch(BridgedConfig.Profile,
* String, String)}, and the launch's resolved agent-session id is returned alongside the agent
* so the caller can put it on the {@link PeerHandle}.
*/
protected Spawned spawnInternal(String profileName, String requestedCwd, String callerCwd,
String sessionName, String resumeSessionId, MemberRole role) {
BridgedConfig.Profile cfg = requireProfile(profileName);
BridgedConfig.Fleet liveFleet = fleet == null ? null : fleet.get();
String roleCharter = liveFleet == null ? null : liveFleet.charterFor(role);
String replyCharter = cfg.hasMcp() ? REPLY_CHARTER : null;
String charter = roleCharter == null ? replyCharter
: replyCharter == null ? roleCharter : roleCharter + "\n\n" + replyCharter;
Launch launch = buildLaunch(cfg, new LaunchSpec(sessionName, resumeSessionId, role, charter));
Launch launch = buildLaunch(cfg, sessionName, resumeSessionId);
String cwd = resolveCwd(requestedCwd, cfg, callerCwd);
Agent agent = cfg.tabPlacement()
? spawnInTab(cfg, launch.env(), launch.argv(), cwd, role, liveFleet)
? spawnInTab(cfg, launch.env(), launch.argv(), cwd, role)
: spawnAsPane(cfg, launch.env(), launch.argv(), cwd);
return new Spawned(agent, launch.agentSessionId());
}
@@ -384,7 +382,7 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
/** Dedicated worker space → own tab (carrying cwd+env) → start the peer into the seed pane. */
private Agent spawnInTab(BridgedConfig.Profile cfg, Map<String, String> workerEnv,
List<String> argv, String cwd, MemberRole role, BridgedConfig.Fleet liveFleet) {
List<String> argv, String cwd, MemberRole role) {
Workspace space = spaces.ensureWorkspace(cfg.workspace());
Tab.Created tab = spaces.createTab(space.workspaceId(), cwd, workerEnv);
log.info("spawning {} profile={} space={} tab={} cwd={}",
@@ -417,7 +415,7 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
tidy("label tab " + tab.tab().tabId(),
() -> spaces.renameTab(tab.tab().tabId(),
cfg.renderTabLabel(
liveFleet == null ? null : liveFleet.tabLabel(),
tabLabelTemplate == null ? null : tabLabelTemplate.get(),
role, nextLabelSeq(role, cfg.profile()))));
log.info("{} started pane={} tab={} terminal={}",
namePrefix, started.agent().paneId(), started.agent().tabId(), started.agent().terminalId());
@@ -38,7 +38,7 @@ import java.util.function.Supplier;
* <li><strong>File-based MCP mount + instructions.</strong> opencode has no inline
* {@code --mcp-config}/{@code --append-system-prompt}. Instead the bridge writes an ephemeral
* {@code opencode.json} that declares the bridge as a {@code remote} MCP server and lists a
* member-charter file under {@code instructions}, then points the worker at it with
* reply-charter file under {@code instructions}, then points the worker at it with
* {@code OPENCODE_CONFIG}. This is the one place the launcher touches disk — Claude never did.</li>
* <li><strong>Model as a flag.</strong> the {@code provider/model} selector is passed as
* {@code -m}, not an env var.</li>
@@ -54,9 +54,38 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
/** Writer for the generated {@code opencode.json}. */
private static final ObjectMapper JSON = new ObjectMapper();
/**
* Standing instruction written to the charter file and mounted via the config's
* {@code instructions} so the worker returns its result through {@code bridge_reply}. Kept on
* disk (not a launch flag) because opencode's {@code instructions} takes file paths, not inline
* text — the file is regenerated per spawn and never touches the worker's own profile.
*/
static final String REPLY_CHARTER =
"You are an off-subscription worker in the claude-bridge fleet, running under opencode. "
+ "Every message you receive arrives through the bridge, and the ONLY channel back to the "
+ "sender is the bridge_reply MCP tool. Text you write in your terminal is NOT sent "
+ "anywhere — the sender cannot see your screen, so an in-terminal answer is silently "
+ "discarded. Therefore you MUST end EVERY turn by calling bridge_reply with `content` set "
+ "to your complete response. This holds for every message without exception — tasks, "
+ "questions, clarifications, acknowledgements, and ordinary back-and-forth conversation. "
+ "Call bridge_reply exactly once, as the final action of your turn, with your full answer "
+ "in `content`; never wait for confirmation first. If you end a turn without calling "
+ "bridge_reply, the sender receives nothing and the exchange stalls.";
/** Root under which per-spawn opencode config dirs are created (injectable for tests). */
private final Path configRoot;
/**
* The current spawn's resume-target session id, threaded from {@link #spawn(SpawnRequest)} to
* {@link #buildLaunch} across the base's {@code spawn -> spawnInternal -> buildLaunch} chain,
* which carries no request. A plain field would race under concurrent spawns (the base supports
* them), so it is thread-local: each spawn captures its own request's id on its own thread, and
* {@code buildLaunch}, synchronous and same-thread, reads exactly that one. Set only around the
* {@code super.spawn} call and cleared in {@code finally}, so a paused/leftover value can never
* bleed into the next spawn.
*/
private final ThreadLocal<String> resumeSessionId = new ThreadLocal<>();
/**
* Session discovery against opencode's on-disk storage ({@link OpenCodeSessionDiscovery}) —
* the one seam that knows opencode's private session-file layout. Its root is injectable for
@@ -97,10 +126,10 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
Function<String, String> env,
long spawnReadyTimeoutMs, long spawnReadyPollMs,
Supplier<BridgedConfig.Fleet> fleet) {
Supplier<String> tabLabelTemplate) {
this(agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs,
System::currentTimeMillis, () -> sleepUninterruptibly(spawnReadyPollMs),
defaultConfigRoot(), defaultDiscoveryRoot(), fleet);
defaultConfigRoot(), defaultDiscoveryRoot(), tabLabelTemplate);
}
/**
@@ -135,7 +164,8 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
/**
* Full testability constructor, plus the fleet-wide tab-label template (CB-557).
*
* @param fleet live fleet config, read once for each spawn
* @param tabLabelTemplate {@code fleet.tabLabel}; {@code null}/blank ⇒
* {@link BridgedConfig.Fleet#DEFAULT_TAB_LABEL}
*/
public OpenCodeLauncher(AgentControl agents, WorkspaceControl spaces,
Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
@@ -143,9 +173,9 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
long spawnReadyTimeoutMs,
LongSupplier nowMillis, Runnable sleeper,
Path configRoot, Path discoveryRoot,
Supplier<BridgedConfig.Fleet> fleet) {
Supplier<String> tabLabelTemplate) {
super(NAME_PREFIX, agents, spaces, profiles, defaultProfile, env,
spawnReadyTimeoutMs, nowMillis, sleeper, fleet);
spawnReadyTimeoutMs, nowMillis, sleeper, tabLabelTemplate);
this.configRoot = configRoot;
this.discovery = new OpenCodeSessionDiscovery(discoveryRoot);
}
@@ -163,21 +193,20 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
* {@inheritDoc}
*
* <p>Builds the opencode launch: no {@code ANTHROPIC_*} and no guard (opencode reads its own
* provider credentials); when the profile mounts the bridge MCP or has a member charter,
* generate an ephemeral {@code opencode.json} (remote MCP server + member-charter instructions)
* and point the worker at it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge
* grant; and select the model with {@code -m}.
* provider credentials); when the profile mounts the bridge MCP, generate an ephemeral
* {@code opencode.json} (remote MCP server + reply-charter instructions) and point the worker at
* it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge grant; and select the model
* with {@code -m}.
*/
@Override
protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {
protected Launch buildLaunch(BridgedConfig.Profile cfg) {
Map<String, String> workerEnv = baseEnv(cfg);
// A config file is needed for the bridge MCP mount, a member charter, or a pinned endpoint (CB-508).
if (cfg.hasMcp() || spec.charter() != null || hasCustomProvider(cfg)) {
workerEnv.put("OPENCODE_CONFIG", writeConfig(cfg, spec.charter()).toString());
// A config file is needed for the bridge MCP mount, for a pinned endpoint (CB-508), or both.
if (cfg.hasMcp() || hasCustomProvider(cfg)) {
workerEnv.put("OPENCODE_CONFIG", writeConfig(cfg).toString());
}
applyGitToken(workerEnv, cfg);
return new Launch(workerEnv,
argvWithResume(argvWithModel(argvWithAuto(cfg), cfg), spec.resumeSessionId()));
return new Launch(workerEnv, argvWithResume(argvWithModel(argvWithAuto(cfg), cfg)));
}
/**
@@ -214,9 +243,11 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
* The launch argv plus, on a resumed spawn, opencode's {@code -s <id>} flag to continue a prior
* conversation by its session id. {@code -s, --session <id>} resumes an existing session; on a
* fresh spawn (no resume target) no flag is added, letting opencode start a brand-new session.
* The id comes from the base launch spec.
* The id comes from the current spawn request's {@code resumeSessionId}, threaded per-thread by
* {@link #spawn(SpawnRequest)}.
*/
private List<String> argvWithResume(List<String> argv, String id) {
private List<String> argvWithResume(List<String> argv) {
String id = resumeSessionId.get();
if (id == null || id.isBlank()) {
return argv;
}
@@ -236,12 +267,12 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
}
/**
* Write an ephemeral {@code opencode.json} (and the member-charter file it references) into a
* Write an ephemeral {@code opencode.json} (and the reply-charter file it references) into a
* fresh per-spawn directory under {@link #configRoot}, and return the config file's path for
* {@code OPENCODE_CONFIG}. The dir is unique per spawn so concurrent workers never race on it;
* it is best-effort cleaned on JVM exit (worker config is disposable — regenerated every spawn).
*/
private Path writeConfig(BridgedConfig.Profile cfg, String charterText) {
private Path writeConfig(BridgedConfig.Profile cfg) {
try {
Path dir = Files.createTempDirectory(configRoot, "bridged-opencode-");
dir.toFile().deleteOnExit();
@@ -264,19 +295,16 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
// if per-profile control is ever wanted, add a profile knob rather than dropping this.
root.putObject("compaction").put("auto", true);
if (charterText != null) {
Path charter = dir.resolve("member-charter.md");
Files.writeString(charter, charterText);
if (cfg.hasMcp()) {
Path charter = dir.resolve("reply-charter.md");
Files.writeString(charter, REPLY_CHARTER);
charter.toFile().deleteOnExit();
root.putArray("instructions").add(charter.toAbsolutePath().toString());
}
if (cfg.hasMcp()) {
ObjectNode bridge = root.putObject("mcp").putObject("bridge");
bridge.put("type", "remote");
bridge.put("url", cfg.mcpUrl());
bridge.put("enabled", true);
root.putArray("instructions").add(charter.toAbsolutePath().toString());
}
if (hasCustomProvider(cfg)) {
addCustomProvider(root, cfg);
@@ -352,11 +380,31 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
return afterScheme.contains("/") ? trimmed : trimmed + "/v1";
}
/** Add lazy on-disk session discovery to the base handle. */
/**
* {@inheritDoc}
*
* <p>adds this adapter's session-identity work around the base's spawn — as opencode cannot be
* told its session id at spawn (see {@link Capability#SESSION_RESUME} vs
* {@link Capability#SESSION_NAME}), identity is only ever adopted after the fact:
* <ul>
* <li>the request's {@code resumeSessionId} is remembered for {@link #buildLaunch} to turn
* into {@code -s <id>}; and</li>
* <li>the returned handle is wrapped so its
* {@link dev.ltms.bridged.peer.PeerHandle#agentSessionId()} performs lazy session
* discovery against opencode's storage (see {@link OpenCodeSessionDiscovery}) — always
* non-blocking, {@code null} until opencode has persisted the session record.</li>
* </ul>
*/
@Override
public PeerHandle spawn(SpawnRequest req) {
PeerHandle inner = super.spawn(req);
return new SessionAwareHandle(inner, discovery, effectiveCwd(req));
resumeSessionId.set(req.resumeSessionId());
try {
PeerHandle inner = super.spawn(req);
return new SessionAwareHandle(inner, discovery, effectiveCwd(req));
} finally {
// Never let a paused/leftover resume id bleed into the next spawn on this thread.
resumeSessionId.remove();
}
}
/**
@@ -69,14 +69,6 @@ public final class MessageService {
* failure context (e.g. the error screen). Terminal, but not a successful completion.
*/
WORKER_FAILED,
/**
* The turn finished without a {@code bridge_reply} and the scrape matched the backend's
* configured usage-limit refusal pattern (CB-578 stage A); {@code text} is the reason,
* carrying the matched line. The worker's pane is healthy — only its account is refusing —
* so this is never reported as a completed reply, and is kept distinct from
* {@link #WORKER_FAILED} (a wedged worker) and a session simply going {@code GONE}.
*/
BACKEND_EXHAUSTED,
/**
* The worker paused mid-turn to ask the primary a question (CB-205); {@code text} is the
* question and {@code turnId} correlates the answer. Not terminal — the primary answers with
@@ -135,8 +127,6 @@ public final class MessageService {
public enum Phase {
/** Delegated and in flight — queued for the worker or being worked. */
PENDING,
/** The worker is paused in {@code bridge_ask}; {@link TaskView#reply} and {@link TaskView#turnId} identify it. */
ASKING,
/** The worker's turn finished; {@link TaskView#reply} holds the answer. */
DONE,
/** The delegation could not complete (timed out, worker gone, or busy). */
@@ -146,28 +136,16 @@ public final class MessageService {
/**
* A poll snapshot of an async delegation.
*
* @param reply the answer when {@link #phase} is {@link Phase#DONE}, or the question when
* {@link #phase} is {@link Phase#ASKING}; otherwise {@code null}
* @param reply the answer when {@link #phase} is {@link Phase#DONE}, else {@code null}
* @param replySource {@code "reply"} (structured {@code bridge_reply}) or {@code "transcript"}
* (completion scrape) when {@link Phase#DONE}, else {@code null}
* @param detail a human note (live worker status while pending, ask state, or failure reason)
* @param turnId correlation id for an {@link Phase#ASKING} ticket, else {@code null}
* @param detail a human note (live worker status while pending, or the failure reason)
*/
public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail,
String turnId) {
public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail) {
}
/** An in-flight or finished async delegation, keyed by its ticket. */
private static final class Task {
private final String target;
private final CompletableFuture<Reply> future = new CompletableFuture<>();
private final long createdNanos = System.nanoTime();
private volatile Reply question;
private volatile String turnId;
private Task(String target) {
this.target = target;
}
private record Task(String target, CompletableFuture<Reply> future, long createdNanos) {
}
private final AgentControl agents;
@@ -178,11 +156,6 @@ public final class MessageService {
private final Metrics metrics; // CB-502: nullable — no registry in unit tests
private final ConcurrentHashMap<String, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
/** Async task that owns each exact forward rendezvous waiter. */
private final ConcurrentHashMap<CompletableFuture<Rendezvous.Resolution>, Task> asyncTasksByWaiter =
new ConcurrentHashMap<>();
/** Async tickets paused on a specific {@code bridge_ask} turn. */
private final ConcurrentHashMap<String, Task> asyncTasksByTurn = new ConcurrentHashMap<>();
private final AtomicLong ticketSeq = new AtomicLong();
private final ExecutorService asyncExecutor = Executors.newThreadPerTaskExecutor(
Thread.ofVirtual().name("bridge-async-", 0).factory());
@@ -229,16 +202,6 @@ public final class MessageService {
return agents.status(target);
}
/** Read-only delegation fact for fleet views. */
public boolean hasAcceptedDelivery(String target) {
return rendezvous.isWaiting(target);
}
/** Read-only inbox fact for fleet views. */
public boolean hasInboxMessage(String target) {
return !inbox.peek(target).isEmpty();
}
/**
* Route a worker's explicit {@code bridge_reply}: resolve an open send, or queue it in the
* inbox if no send is currently open. Unlike the bare {@link Rendezvous#resolve}, a no-waiter
@@ -288,7 +251,6 @@ public final class MessageService {
case COMPLETED_UNREPLIED -> "completion_fallback";
case TIMED_OUT_WORKING, TIMED_OUT_QUEUED, BUSY -> "timeout";
case WORKER_FAILED -> "failed";
case BACKEND_EXHAUSTED -> "backend_exhausted";
case STALE_TURN, QUESTION -> null; // not a completed delegation
};
}
@@ -310,18 +272,14 @@ public final class MessageService {
*/
public boolean abandon(String target, String reason) {
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason);
boolean asyncFailed = false;
for (Task task : tasks.values()) {
if (target.equals(task.target) && task.question == null
&& task.future.complete(new Reply(Outcome.WORKER_FAILED, reason))) {
asyncFailed = true;
}
if (waiter == null || waiter.isDone()) {
return false; // nobody is blocked on this worker — nothing to abandon
}
boolean failed = rendezvous.resolveFailure(waiter, reason);
if (failed) {
log.warn("abandoning the blocked send to {}: {}", target, reason);
}
return failed || asyncFailed;
return failed;
}
/**
@@ -368,11 +326,6 @@ public final class MessageService {
* never earned. {@code null} disables the hook.
*/
public Reply send(String target, String content, long timeoutMillis, Runnable onAccepted) {
return send(target, content, timeoutMillis, onAccepted, null);
}
/** Run a send, optionally stopping an async task that teardown already failed before acceptance. */
private Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, Task task) {
long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L;
ReentrantLock lock = sessionLocks.computeIfAbsent(target, _ -> new ReentrantLock());
@@ -380,12 +333,6 @@ public final class MessageService {
return new Reply(Outcome.BUSY, null); // another send held the session the whole window
}
try {
if (task != null && task.future.isDone()) {
return task.future.getNow(null);
}
if (hasAsyncQuestion(target)) {
return new Reply(Outcome.BUSY, null); // the worker's current turn is paused for its lead
}
// Open the waiter BEFORE queueing delivery (CB-548). A fast reply — the worker already
// injectable the instant we enqueue — otherwise arrives before the waiter is registered
// and orphans into the inbox while this send blocks to the timeout (the enqueue-before-
@@ -394,17 +341,13 @@ public final class MessageService {
// failed send leaves no stale waiter behind.
CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
try {
if (task != null) {
asyncTasksByWaiter.put(reply, task);
}
TurnToken token = new TurnToken(target, reply);
// The send has won the lock; the accepted-delivery hook records delegator ownership
// here (CB-548). It runs BEFORE enqueue so a throwing hook — onAccepted is now a
// public callback — fails the send without queuing a message that would orphan.
if (onAccepted != null) {
onAccepted.run();
}
CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
CompletableFuture<Void> delivered = injector.enqueue(target, content);
try {
Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
@@ -421,7 +364,6 @@ public final class MessageService {
throw new IllegalStateException("interrupted awaiting reply from " + target, e);
}
} finally {
asyncTasksByWaiter.remove(reply);
rendezvous.close(target, reply);
}
} finally {
@@ -447,12 +389,7 @@ public final class MessageService {
if (ticket.fresh()) {
// Register the reverse waiter first, then surface the question — so the answer, which can
// arrive the instant the primary reacts, always finds an open waiter to resolve.
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(workerSession);
Task task = markAsyncQuestion(waiter, question, ticket.turnId());
if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
if (task != null) {
clearAsyncQuestion(ticket.turnId(), true);
}
rendezvous.closeAsk(ticket.turnId());
return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
}
@@ -462,7 +399,6 @@ public final class MessageService {
return new AskResult(AskOutcome.ANSWERED, answer);
} catch (TimeoutException e) {
log.debug("bridge_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
clearAsyncQuestion(ticket.turnId(), true);
return new AskResult(AskOutcome.TIMED_OUT, null);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
@@ -505,12 +441,9 @@ public final class MessageService {
rendezvous.close(workerSession, reply);
return new Reply(Outcome.STALE_TURN, null); // lapsed between the lookup and the unblock
}
clearAsyncQuestion(turnId, false);
try {
Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
Reply result = new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
finishAsyncTask(turnId, result);
return result;
return new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
} catch (TimeoutException e) {
// The worker resumed but hasn't replied yet — no completion fallback arms an answered
// turn (it never re-entered the injector), so a silent worker rides out the window.
@@ -550,21 +483,9 @@ public final class MessageService {
*/
public String sendAsync(String target, String content, Runnable onAccepted) {
String ticket = "task-" + ticketSeq.incrementAndGet();
Task task = new Task(target);
tasks.put(ticket, task);
asyncExecutor.submit(() -> {
try {
Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task);
if (result.outcome() == Outcome.QUESTION) {
// Keep the accepted owner until answer() finishes it. markAsyncQuestion may run
// just after resolveQuestion wakes this thread.
} else {
finishAsyncTask(task, result);
}
} catch (Throwable t) {
task.future.completeExceptionally(t);
}
});
CompletableFuture<Reply> future = CompletableFuture.supplyAsync(
() -> send(target, content, ASYNC_TIMEOUT_MS, onAccepted), asyncExecutor);
tasks.put(ticket, new Task(target, future, System.nanoTime()));
pruneTerminalTickets();
log.debug("async send {} -> {}", ticket, target);
return ticket;
@@ -580,34 +501,27 @@ public final class MessageService {
if (task == null) {
return null;
}
CompletableFuture<Reply> f = task.future;
CompletableFuture<Reply> f = task.future();
if (!f.isDone()) {
Reply question = task.question;
if (question != null) {
return new TaskView(ticket, Phase.ASKING, question.text(), null,
"worker is waiting for your answer", question.turnId());
}
return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target), null);
return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target()));
}
Reply r;
try {
r = f.getNow(null);
} catch (CompletionException | java.util.concurrent.CancellationException e) {
Throwable cause = (e instanceof CompletionException ce && ce.getCause() != null) ? ce.getCause() : e;
return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage(), null);
return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage());
}
if (r.completed()) {
String source = r.outcome() == Outcome.REPLIED ? "reply" : "transcript";
return new TaskView(ticket, Phase.DONE, r.text(), source, null, null);
return new TaskView(ticket, Phase.DONE, r.text(), source, null);
}
// A wedged worker (CB-109) or a backend-exhausted classification (CB-578 stage A) carries
// the real cause as its reason; the timeout/busy outcomes carry none, so fall back to the
// outcome name.
boolean carriesReason = r.outcome() == Outcome.WORKER_FAILED || r.outcome() == Outcome.BACKEND_EXHAUSTED;
String detail = carriesReason && r.text() != null
// A wedged worker (CB-109) carries the error context as its reason; the timeout/busy
// outcomes carry none, so fall back to the outcome name.
String detail = r.outcome() == Outcome.WORKER_FAILED && r.text() != null
? r.text()
: "no reply — " + r.outcome().name().toLowerCase();
return new TaskView(ticket, Phase.FAILED, null, null, detail, null);
return new TaskView(ticket, Phase.FAILED, null, null, detail);
}
/** Best-effort live worker status for a pending poll; never throws (a lookup error is just noise). */
@@ -622,51 +536,7 @@ public final class MessageService {
/** Drop finished tickets older than the TTL so the registry cannot grow without bound. */
private void pruneTerminalTickets() {
long cutoff = System.nanoTime() - TICKET_TTL_NANOS;
tasks.values().removeIf(t -> t.future.isDone() && t.createdNanos < cutoff);
}
/** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */
private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
if (task != null) {
task.question = new Reply(Outcome.QUESTION, text, turnId);
task.turnId = turnId;
asyncTasksByTurn.put(turnId, task);
}
return task;
}
/** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
Task task = asyncTasksByTurn.get(turnId);
if (task != null && turnId.equals(task.turnId)) {
task.question = null;
if (forgetTurn) {
asyncTasksByTurn.remove(turnId, task);
task.turnId = null;
}
}
}
/** Complete and detach an async ticket after its worker's actual terminal reply. */
private void finishAsyncTask(Task task, Reply result) {
task.future.complete(result);
if (task.turnId != null) {
asyncTasksByTurn.remove(task.turnId, task);
}
}
/** Complete the async ticket correlated to a specific answered turn. */
private void finishAsyncTask(String turnId, Reply result) {
Task task = asyncTasksByTurn.get(turnId);
if (task != null) {
finishAsyncTask(task, result);
}
}
/** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
private boolean hasAsyncQuestion(String target) {
return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
tasks.values().removeIf(t -> t.future().isDone() && t.createdNanos() < cutoff);
}
/** Release the async executor. */
@@ -680,7 +550,6 @@ public final class MessageService {
case REPLY -> Outcome.REPLIED;
case COMPLETION -> Outcome.COMPLETED_UNREPLIED;
case FAILED -> Outcome.WORKER_FAILED;
case BACKEND_EXHAUSTED -> Outcome.BACKEND_EXHAUSTED;
case QUESTION -> Outcome.QUESTION;
};
}
@@ -33,13 +33,6 @@ public final class Rendezvous {
COMPLETION,
/** The worker ran the turn then wedged (CB-109); {@code text} is the failure context. */
FAILED,
/**
* The turn finished without a {@code bridge_reply}, and the scrape matched the backend's
* configured usage-limit refusal pattern (CB-578 stage A); {@code text} is the reason,
* carrying the matched line. The pane is healthy — only the account is refusing — so this
* is kept separate from a session simply going {@code GONE}.
*/
BACKEND_EXHAUSTED,
/**
* The worker paused mid-turn to ask the primary a question (CB-205 reverse rendezvous);
* {@code text} is the question and {@code turnId} correlates the primary's answer back to
@@ -231,19 +224,6 @@ public final class Rendezvous {
return waiter != null && waiter.complete(new Resolution(Kind.FAILED, reason));
}
/**
* Resolve a specific captured {@code waiter} as {@link Kind#BACKEND_EXHAUSTED} (CB-578 stage A):
* the turn finished with no {@code bridge_reply} and the scrape matched the backend's configured
* usage-limit pattern; {@code reason} carries the matched line. Like
* {@link #resolveCompletion(CompletableFuture, String)} it targets the exact captured send
* (CB-116). A no-op if that waiter was already resolved — first resolution wins.
*
* @return {@code true} if this call resolved the waiter, {@code false} if it was null or already resolved
*/
public boolean resolveExhausted(CompletableFuture<Resolution> waiter, String reason) {
return waiter != null && waiter.complete(new Resolution(Kind.BACKEND_EXHAUSTED, reason));
}
private boolean complete(String session, Resolution resolution) {
CompletableFuture<Resolution> waiter = waiters.get(session);
return waiter != null && waiter.complete(resolution);
@@ -1,20 +0,0 @@
package dev.ltms.bridged.msg;
import java.util.concurrent.CompletableFuture;
/**
* Identity for one accepted send. The session turn is deliberately absent: CompletionResolver's
* delivery callback runs before SessionManager.onDelivered, so binding it needs a later ordering design.
*/
public final class TurnToken {
private final String target;
private final CompletableFuture<Rendezvous.Resolution> waiter;
public TurnToken(String target, CompletableFuture<Rendezvous.Resolution> waiter) {
this.target = target;
this.waiter = waiter;
}
public String target() { return target; }
public CompletableFuture<Rendezvous.Resolution> waiter() { return waiter; }
}
@@ -403,12 +403,9 @@ public final class BridgedApp {
case TIMED_OUT_QUEUED -> "queued";
case BUSY -> "busy";
case WORKER_FAILED -> "failed";
case BACKEND_EXHAUSTED -> "backend_exhausted";
default -> "done"; // unreachable (terminal outcomes handled above)
},
"detail", (reply.outcome() == MessageService.Outcome.WORKER_FAILED
|| reply.outcome() == MessageService.Outcome.BACKEND_EXHAUSTED)
&& reply.text() != null
"detail", reply.outcome() == MessageService.Outcome.WORKER_FAILED && reply.text() != null
? reply.text()
: "no reply within " + timeout + "ms; poll status or retry"));
}
@@ -167,22 +167,6 @@ public final class GitWorktrees implements Worktrees {
exec("git", "-C", repoRoot, "worktree", "remove", "--force", worktreePath);
}
@Override
public boolean hasUncommitted(String worktreePath) {
// A worktree that is already gone holds no work to lose, and it must not break teardown:
// git -C <missing-dir> status exits non-zero and would throw where release() is mid-way
// through stopping a pane. Mirror remove()'s already-gone tolerance by treating it as clean.
Path p = Path.of(worktreePath);
if (!Files.exists(p)) {
log.debug("worktree {} already gone — nothing can be uncommitted", worktreePath);
return false;
}
// No --untracked-files=no: the exact shape of the work lost in CB-576 was a new file
// that was never added, so an untracked-only worktree is still dirty.
String out = exec("git", "-C", worktreePath, "status", "--porcelain");
return !out.isBlank();
}
@Override
public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
if (overlay == null || overlay.isEmpty()) {
@@ -4,7 +4,6 @@ import dev.ltms.bridged.auth.MemberLifecycle;
import dev.ltms.bridged.herdr.Agent;
import dev.ltms.bridged.inject.TurnListener;
import dev.ltms.bridged.inject.MemberPresence;
import dev.ltms.bridged.msg.TurnToken;
import dev.ltms.bridged.peer.MemberRole;
import dev.ltms.bridged.peer.PeerHandle;
import dev.ltms.bridged.peer.PeerLauncher;
@@ -201,16 +200,6 @@ public final class SessionManager implements TurnListener {
removed.paneId(), removed.terminalId(), removed.state(), cause);
if (preserveWorktree && removed.worktree() != null) {
logPreservedForShutdown(removed);
} else if (removed.worktree() != null && worktrees.hasUncommitted(removed.worktree())) {
// CB-576: a release that would otherwise remove the worktree finds it holding
// uncommitted work the bridge cannot see. A worker that ends a turn without
// committing (normally because it stopped to ask a question or refused the turn)
// has its only copy of that work in the worktree. Remove would --force-delete it,
// so preserve the directory and tell an operator where to find it.
preserveWorktree = true;
log.warn("release {} preserves dirty worktree {} for pane={} terminal={}: "
+ "the worktree holds uncommitted changes that --force remove would destroy",
cause, removed.worktree(), removed.paneId(), removed.terminalId());
}
// CB-516: a send still waiting on this worker can never be answered now. Tell the
// listener BEFORE the pane is torn down, so a blocked caller fails fast with a real
@@ -436,7 +425,7 @@ public final class SessionManager implements TurnListener {
* can be re-delivered for multi-turn reuse until it is released.
*/
@Override
public void onDelivered(String target, TurnToken token) {
public void onDelivered(String target) {
MemberSession current = findByTerminal(target);
if (current == null) return;
if (current.state() != MemberSession.State.READY && current.state() != MemberSession.State.DONE) {
@@ -10,18 +10,6 @@ public interface Worktrees {
/** git -C <repoRoot> worktree remove --force <path>. Idempotent (already-gone tolerated). */
void remove(String repoRoot, String worktreePath);
/**
* True when the worktree holds uncommitted changes the bridge cannot see: tracked
* modifications, staged files, or untracked files. {@code git status --porcelain} is the
* test; an empty result means clean. Callers use this to decide whether removing the
* worktree would silently destroy a worker's only copy of its work.
*
* <p>An already-gone worktree is reported as clean (no throw), matching {@link #remove}'s
* idempotent contract: a path that does not exist holds no work to lose, and must not break
* a teardown that is mid-way through stopping the pane.
*/
boolean hasUncommitted(String worktreePath);
/** Copy each existing overlay path repoRoot→worktree; mark tracked ones --skip-worktree. */
void overlayParity(String repoRoot, String worktreePath, List<String> overlay);
-9
View File
@@ -1,13 +1,4 @@
<configuration>
<!--
CB-575: MCP SDK 2.0.0 has no public notification registration API. It registers only
notifications/initialized and notifications/roots/list_changed, so other client notifications
still warn when unhandled. Clients may legitimately send notifications/cancelled; suppress only
that SDK WARN because it would devalue the action-needed WARN level used by M4 fleet health.
If a later SDK handles cancellation, this filter simply stops matching and can be removed.
-->
<turboFilter class="dev.ltms.bridged.logging.McpCancelledNotificationFilter"/>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} %-5level [%thread] %logger{28} - %msg%n</pattern>
@@ -194,7 +194,7 @@ class BridgedConfigTest {
fleet:
leaders:
opus:
tab: "lead: opus"
terminal: term_opus
""");
BridgedConfig.Leader lead = BridgedConfig.load(f).fleet().leaders().get("opus");
@@ -212,7 +212,7 @@ class BridgedConfigTest {
fleet:
leaders:
opus:
tab: "drive: opus"
terminal: term_opus
tabPrefix: "drive:"
scanIntervalSeconds: 30
""");
@@ -224,9 +224,7 @@ class BridgedConfigTest {
/**
* The pane no longer has to exist before the daemon does (CB-557): a lead naming a profile may
* be launched, while one that names no profile is recognised and never created. Either way it
* still needs its own {@code tab:} (CB-579) — that part is unconditional, see
* {@link #aLeadWithNoTabRefusesToStart}.
* be launched, while one that names only a terminal is recognised and never created.
*/
@Test
void aLeadIsCreatableOnlyWhenItNamesAProfile(@TempDir Path dir) throws Exception {
@@ -241,9 +239,8 @@ class BridgedConfigTest {
leaders:
launched:
profile: opus
tab: "lead: launched"
pinned:
tab: "lead: pinned"
terminal: term_opus
""");
var leaders = BridgedConfig.load(f).fleet().leaders();
@@ -252,12 +249,8 @@ class BridgedConfigTest {
"no profile to launch on ⇒ recognise-only, the pre-CB-557 behaviour");
}
/**
* CB-579: {@code tab} is the only field a lead's identity depends on now, so it is required
* whether the entry is creatable or recognise-only — without it the entry can never be found.
*/
@Test
void aLeadWithNoTabRefusesToStart(@TempDir Path dir) throws Exception {
void aLeadThatCanBeNeitherFoundNorCreatedRefusesToStart(@TempDir Path dir) throws Exception {
Path f = dir.resolve("useless-lead.yaml");
Files.writeString(f, """
bind:
@@ -271,80 +264,6 @@ class BridgedConfigTest {
IllegalStateException e = assertThrows(IllegalStateException.class, cfg::validateMembers);
assertTrue(e.getMessage().contains("ghost"), "the message must name the useless entry");
assertTrue(e.getMessage().contains("tab:"), "the message must say what is missing");
}
/**
* CB-579 acceptance (2): a config still spelling {@code fleet.leaders.<name>.terminal} must fail
* loudly at load, not be silently dropped by {@code Leader}'s {@code @JsonIgnoreProperties}.
*/
@Test
void aLeaderTerminalKeyFailsLoadAndNamesTabAsTheReplacement(@TempDir Path dir) throws Exception {
Path f = dir.resolve("stale-terminal.yaml");
Files.writeString(f, """
bind:
port: 8080
fleet:
leaders:
opus:
terminal: term_opus
""");
IllegalStateException e =
assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
assertTrue(e.getMessage().contains("opus"), "the message must name the offending entry");
assertTrue(e.getMessage().contains("tab:"), "the message must name the replacement key");
assertTrue(e.getMessage().contains("terminal"), "the message must name the retired key");
}
/** The same refusal, and it must name every offending entry, not just the first. */
@Test
void everyLeaderStillUsingTerminalIsReportedAtOnce(@TempDir Path dir) throws Exception {
Path f = dir.resolve("stale-terminals.yaml");
Files.writeString(f, """
bind:
port: 8080
fleet:
leaders:
opus:
terminal: term_opus
sol:
terminal: term_sol
""");
IllegalStateException e =
assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
assertTrue(e.getMessage().contains("opus"));
assertTrue(e.getMessage().contains("sol"));
}
/** A {@code terminal:} anywhere else in the document (not under a leader entry) is unaffected. */
@Test
void aTerminalKeyOutsideFleetLeadersIsNotRejected(@TempDir Path dir) throws Exception {
Path f = dir.resolve("primary-terminal-ok.yaml");
Files.writeString(f, "bind:\n port: 8080\nprimary:\n terminal: term_fixed\n");
assertDoesNotThrow(() -> BridgedConfig.load(f));
}
/** CB-579 acceptance (3): distinct `tab:` labels need no shared prefix — one scanner finds both. */
@Test
void twoLeadersWithDifferentTabsAreBothConfigured(@TempDir Path dir) throws Exception {
Path f = dir.resolve("two-tabs.yaml");
Files.writeString(f, """
bind:
port: 8080
fleet:
leaders:
opus:
tab: "lead: opus"
sol:
tab: "captain: sol"
""");
var leaders = BridgedConfig.load(f).fleet().leaders();
assertEquals("lead: opus", leaders.get("opus").tab());
assertEquals("captain: sol", leaders.get("sol").tab());
}
// ── CB-551: the idle-lead heartbeat ─────────────────────────────────────────────────────────
@@ -409,7 +328,7 @@ class BridgedConfigTest {
fleet:
leaders:
opus:
tab: "lead: opus"
terminal: term_opus
tabPrefix: "lead:"
""");
BridgedConfig cfg = BridgedConfig.load(f);
@@ -430,7 +349,7 @@ class BridgedConfigTest {
tabLabel: "lead: {role} {profile}"
leaders:
opus:
tab: "lead: opus"
terminal: term_opus
""");
BridgedConfig cfg = BridgedConfig.load(f);
@@ -455,7 +374,7 @@ class BridgedConfigTest {
fleet:
leaders:
opus:
tab: "lead: opus"
terminal: term_opus
""");
assertDoesNotThrow(() -> BridgedConfig.load(f).validateLeadTabPrefixes());
@@ -481,7 +400,7 @@ class BridgedConfigTest {
"a label that collides with a convention nobody reads is not a problem");
}
// ── CB-530/CB-579: the leaders registry ─────────────────────────────────────────────────────
// ── CB-530: the leaders registry ────────────────────────────────────────────────────────────
@Test
void leadersBlockRegistersEveryPaneByName(@TempDir Path dir) throws Exception {
@@ -492,10 +411,10 @@ class BridgedConfigTest {
fleet:
leaders:
opus-5.0:
tab: "lead: opus-5.0"
terminal: term_opus
kind: claude
gpt-sol-5.6:
tab: "lead: gpt-sol-5.6"
terminal: term_sol
kind: opencode
model: openai/gpt-5.6-terra
""");
@@ -506,9 +425,9 @@ class BridgedConfigTest {
assertEquals(Set.of("opus-5.0", "gpt-sol-5.6"), leaders.keySet());
assertEquals("opencode", leaders.get("gpt-sol-5.6").kind());
assertEquals("openai/gpt-5.6-terra", leaders.get("gpt-sol-5.6").model());
// Identity is the tab now (CB-579) — both entries carry their own, distinct label.
assertEquals("lead: opus-5.0", leaders.get("opus-5.0").tab());
assertEquals("lead: gpt-sol-5.6", leaders.get("gpt-sol-5.6").tab());
// The whole point: BOTH panes resolve as leads, so neither is demoted to worker.
assertEquals(Map.of("term_opus", "opus-5.0", "term_sol", "gpt-sol-5.6"),
cfg.leaderTerminals());
}
@Test
@@ -520,6 +439,42 @@ class BridgedConfigTest {
"configs that never migrate must behave exactly as they did before CB-530");
}
@Test
void anExplicitLeadersEntryWinsOverThePinForTheSameTerminal(@TempDir Path dir) throws Exception {
Path f = dir.resolve("both.yaml");
Files.writeString(f, """
bind:
port: 8080
primary:
terminal: term_shared
fleet:
leaders:
opus-5.0:
terminal: term_shared
""");
assertEquals(Map.of("term_shared", "opus-5.0"), BridgedConfig.load(f).leaderTerminals(),
"the pin is the older spelling of the same fact; the named entry is what was meant");
}
@Test
void bothBlocksTogetherRegisterTheUnionOfTheirTerminals(@TempDir Path dir) throws Exception {
Path f = dir.resolve("union.yaml");
Files.writeString(f, """
bind:
port: 8080
primary:
terminal: term_pinned
fleet:
leaders:
gpt-sol-5.6:
terminal: term_sol
""");
assertEquals(Map.of("term_pinned", "primary", "term_sol", "gpt-sol-5.6"),
BridgedConfig.load(f).leaderTerminals());
}
@Test
void neitherBlockLeavesNothingRegistered(@TempDir Path dir) throws Exception {
Path f = dir.resolve("none.yaml");
@@ -528,25 +483,22 @@ class BridgedConfigTest {
assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty());
}
/**
* CB-579: {@code fleet.leaders} no longer feeds {@code leaderTerminals()} at all — a lead's
* identity comes from the live tab scan, not a config-held terminal map. This method now exists
* only for the {@code primary.terminal} fallback.
*/
/** A lead entry with no terminal identifies nothing — it must not register a null key. */
@Test
void fleetLeadersNeverContributesToLeaderTerminals(@TempDir Path dir) throws Exception {
Path f = dir.resolve("leaders-only.yaml");
void aLeadWithoutATerminalIsNotRegistered(@TempDir Path dir) throws Exception {
Path f = dir.resolve("no-terminal.yaml");
Files.writeString(f, """
bind:
port: 8080
fleet:
leaders:
opus-5.0:
tab: "lead: opus-5.0"
sketch:
kind: opencode
real:
terminal: term_real
""");
assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty(),
"no primary.terminal pin ⇒ nothing registered, even with fleet.leaders configured");
assertEquals(Map.of("term_real", "real"), BridgedConfig.load(f).leaderTerminals());
}
// ── CB-548: the architects registry ────────────────────────────────────────────────────────
@@ -1140,7 +1092,6 @@ class BridgedConfigTest {
gitHostEnv: GITEA_HOST
weight: 0.5
maxLoad: 2
exhaustedPattern: "usage limit has been reached"
placement: weighted
lifecycle:
idleTtlSeconds: 300
@@ -1172,8 +1123,6 @@ class BridgedConfigTest {
assertEquals("GITEA_HOST", w.gitHostEnv());
assertEquals(0.5f, w.weight(), 0.0001f, "weight binds as a float");
assertEquals(2, w.maxLoad(), "maxLoad binds as an integer");
assertTrue(w.hasExhaustedPattern(), "exhaustedPattern binds and enables the CB-578 stage A classification");
assertEquals("usage limit has been reached", w.exhaustedPattern());
assertEquals("weighted", cfg.placement(), "placement binds at the top level");
assertEquals(300, cfg.lifecycle().idleTtlSeconds());
@@ -1,170 +0,0 @@
package dev.ltms.bridged.health;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.inject.Injector;
import dev.ltms.bridged.msg.InMemoryReplyInbox;
import dev.ltms.bridged.msg.MessageService;
import dev.ltms.bridged.msg.Rendezvous;
import dev.ltms.bridged.session.SessionManager;
import dev.ltms.bridged.member.ClaudeCodeLauncher;
import dev.ltms.bridged.herdr.WorkspaceControl;
import dev.ltms.bridged.guard.SubscriptionGuard;
import dev.ltms.bridged.config.BridgedConfig;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
class FleetHealthMonitorTest {
@Test void oneTickUsesOneFleetListForAnyRosterSize() {
FakeHerdr herdr = new FakeHerdr();
AgentControl agents = new AgentControl(herdr);
BridgedConfig.Profile profile = new BridgedConfig.Profile("test", "http://test:1", null,
null, null, null, null, null, null, null, null, null);
ClaudeCodeLauncher launcher = new ClaudeCodeLauncher(agents, new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("test")), Map.of("test", profile), "test", _ -> "token");
SessionManager sessions = new SessionManager(launcher);
sessions.acquire("test", null, null, null);
sessions.acquire("test", null, null, null);
herdr.calls.clear();
MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox());
var scheduler = Executors.newSingleThreadScheduledExecutor();
FleetHealthMonitor monitor = new FleetHealthMonitor(agents, sessions::roster, messages, scheduler, () -> 1, 60,
(_, _) -> { });
monitor.tick();
monitor.stop();
assertEquals(1, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
}
@Test void failedTickDoesNotStopTheNextTick() {
FakeHerdr herdr = new FakeHerdr().healthy(false);
AgentControl agents = new AgentControl(herdr);
var scheduler = Executors.newSingleThreadScheduledExecutor();
FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
scheduler, () -> 1, 60, (_, _) -> { });
monitor.tick();
herdr.healthy(true);
monitor.tick();
monitor.stop();
assertEquals(2, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
}
@Test void faultTransitionLogsOnlyOnceUntilItChanges() {
Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
try {
FakeHerdr herdr = new FakeHerdr();
AgentControl agents = new AgentControl(herdr);
var scheduler = Executors.newSingleThreadScheduledExecutor();
FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
scheduler, () -> 1, 60, (_, _) -> { });
monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
monitor.stop();
assertEquals(1, appender.list.stream().filter(event -> event.getFormattedMessage()
.contains("member=term_a state=TURN_BOUNDARY_LOST")).count());
} finally {
logger.detachAppender(appender);
}
}
// --- CB-580: a member that reaches GONE/NEVER_READY must fail its waiting tickets
private static FleetHealthMonitor monitorWith(BiConsumer<String, String> failTarget) {
FakeHerdr herdr = new FakeHerdr();
AgentControl agents = new AgentControl(herdr);
var scheduler = Executors.newSingleThreadScheduledExecutor();
return new FleetHealthMonitor(agents, java.util.List::of,
new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
scheduler, () -> 1, 60, failTarget);
}
@Test void terminalTransitionFailsTheTargetOnce() {
RecordingFailTarget failTarget = new RecordingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.stop();
assertEquals(1, failTarget.calls.size());
assertEquals("term_a", failTarget.calls.get(0).target());
assertTrue(failTarget.calls.get(0).reason().contains("GONE"));
}
@Test void neverReadyNamesItselfAsTheReason() {
RecordingFailTarget failTarget = new RecordingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.NEVER_READY);
monitor.stop();
assertEquals(1, failTarget.calls.size());
assertTrue(failTarget.calls.get(0).reason().contains("NEVER_READY"));
}
@Test void stayingInATerminalStateProducesOneFailureNotN() {
RecordingFailTarget failTarget = new RecordingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.stop();
assertEquals(1, failTarget.calls.size());
}
@Test void aNonTerminalFaultStateDoesNotFailTheTarget() {
RecordingFailTarget failTarget = new RecordingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
monitor.stop();
assertEquals(0, failTarget.calls.size());
}
@Test void failTargetRetryIsBounded() {
AlwaysThrowingFailTarget failTarget = new AlwaysThrowingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.GONE);
monitor.stop();
assertEquals(FleetHealthMonitor.MAX_FAIL_TARGET_ATTEMPTS, failTarget.calls);
}
@Test void exhaustedRetryStillDoesNotRefireOnAnUnchangedTick() {
AlwaysThrowingFailTarget failTarget = new AlwaysThrowingFailTarget();
FleetHealthMonitor monitor = monitorWith(failTarget);
monitor.reportTransition("term_a", HealthState.GONE);
int afterFirstTransition = failTarget.calls;
monitor.reportTransition("term_a", HealthState.GONE);
monitor.stop();
assertEquals(afterFirstTransition, failTarget.calls);
}
private record RecordedCall(String target, String reason) { }
private static final class RecordingFailTarget implements BiConsumer<String, String> {
final java.util.List<RecordedCall> calls = new java.util.ArrayList<>();
@Override public void accept(String target, String reason) {
calls.add(new RecordedCall(target, reason));
}
}
private static final class AlwaysThrowingFailTarget implements BiConsumer<String, String> {
int calls = 0;
@Override public void accept(String target, String reason) {
calls++;
throw new RuntimeException("boom");
}
}
}
@@ -1,56 +0,0 @@
package dev.ltms.bridged.health;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.session.MemberSession;
import dev.ltms.bridged.msg.Rendezvous;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
class FleetHealthTest {
@Test void muteCountsOnlyCompletionFallbacks() {
MuteCounter mute = new MuteCounter();
mute.observe("target", "terra", Rendezvous.Kind.REPLY);
mute.observe("target", "terra", Rendezvous.Kind.COMPLETION);
assertEquals(1, mute.forTarget("target"));
assertEquals(1, mute.forProfile("terra"));
}
@Test void turnBoundaryNeedsTwoSnapshots() {
HealthSnapshot s = snapshot(MemberSession.State.BUSY, AgentStatus.DONE, true);
HealthDecision first = FleetHealth.decide(s, HealthPrior.NONE, 1);
assertEquals(HealthState.WORKING, first.state());
assertEquals(new HealthPrior(true), first.prior());
assertEquals(HealthState.TURN_BOUNDARY_LOST, FleetHealth.decide(s, first.prior(), 2).state());
}
@Test void unknownLiveStatusWithAcceptedDeliveryIsNotIdle() {
assertEquals(HealthState.WORKING, FleetHealth.decide(
snapshot(MemberSession.State.BUSY, AgentStatus.UNKNOWN, true), HealthPrior.NONE, 1).state());
}
@Test void acceptedDeliveryNeverReportsIdle() {
for (MemberSession.State session : MemberSession.State.values()) {
for (AgentStatus live : AgentStatus.values()) {
HealthSnapshot s = snapshot(session, live, true);
assertEquals(false, FleetHealth.decide(s, HealthPrior.NONE, 1).state() == HealthState.IDLE,
() -> "accepted delivery returned IDLE for " + session + "/" + live);
}
}
}
@Test void blockedDoesNotGuessPromptKind() {
assertEquals(HealthState.BLOCKED_AMBIGUOUS, FleetHealth.decide(
snapshot(MemberSession.State.BUSY, AgentStatus.BLOCKED, true), HealthPrior.NONE, 1).state());
}
@Test void controlLinkOutranksMemberFault() {
HealthSnapshot s = new HealthSnapshot(MemberSession.State.BUSY, AgentStatus.DONE, true, false,
false, true, true, true, false, true, true, true);
assertEquals(HealthState.CONTROL_LINK_DOWN, FleetHealth.decide(s, HealthPrior.NONE, 1).state());
}
private static HealthSnapshot snapshot(MemberSession.State state, AgentStatus live, boolean accepted) {
return new HealthSnapshot(state, live, accepted, false, false, true, false, false,
false, false, false, false);
}
}
@@ -1,14 +0,0 @@
package dev.ltms.bridged.health;
import org.junit.jupiter.api.Test;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
class PaneBudgetTest {
@Test void fixedLimitsIgnoreWeakerConfig() {
PaneBudget budget = new PaneBudget();
List<String> targets = List.of("a", "b", "c");
assertEquals(List.of("a", "b"), budget.choose(targets, 0, 0));
assertEquals(List.of("c"), budget.choose(targets, 1, 0));
}
}
@@ -15,9 +15,8 @@ import java.util.concurrent.atomic.AtomicLong;
import static org.junit.jupiter.api.Assertions.*;
/**
* CB-531/CB-579. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that
* an operator-labelled tab matching a configured {@code tab:} is what makes a pane a lead, and —
* just as importantly — what does not, and that a stale entry does not linger forever.
* CB-531. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that an
* operator-labelled tab is what makes a pane a lead, and — just as importantly — what does not.
*/
class LeadTabScannerTest {
@@ -121,48 +120,23 @@ class LeadTabScannerTest {
.pane("w9:p1", "w9:t1", "term_worker");
}
/** The {@code tab:} → name map {@code twoLeads()}'s two lead tabs are configured under. */
private static Map<String, String> twoLeadsConfigured() {
return Map.of("lead: opus-5.0", "opus-5.0", "lead: gpt-sol-5.6", "gpt-sol-5.6");
}
private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> tabToName,
private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> configured,
AtomicLong clock) {
return new LeadTabScanner(herdr, tabToName, Set.of("bridged-workers"), TTL, clock::get);
return new LeadTabScanner(herdr, "lead:", Set.of("bridged-workers"), configured, TTL,
clock::get);
}
@Test
void everyConfiguredTabBecomesALeadNamedByItsEntry() {
Map<String, String> leads = scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong()).get();
void everyLabelledTabBecomesALeadNamedByItsLabel() {
Map<String, String> leads = scanner(twoLeads(), Map.of(), new AtomicLong()).get();
assertEquals(Map.of("term_opus", "opus-5.0", "term_gpt", "gpt-sol-5.6"), leads,
"two leads discovered by their configured tab — no terminal_id was ever configured");
"two leads discovered from labels alone — no terminal_id was ever configured");
}
@Test
void anUnconfiguredTabContributesNothing() {
assertFalse(scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong())
.get().containsKey("term_notes"));
}
/**
* CB-579: matching is exact against the configured map now, not a shared prefix — two leads with
* completely different labels are both discovered by one scanner, no convention required.
*/
@Test
void twoLeadsWithCompletelyDifferentLabelsAreBothDiscovered() {
TopologyHerdr herdr = new TopologyHerdr()
.workspace("w1", "main")
.tab("w1:t1", "w1", "orchestrator: opus")
.tab("w1:t2", "w1", "captain: sol")
.pane("w1:p1", "w1:t1", "term_opus")
.pane("w1:p2", "w1:t2", "term_sol");
Map<String, String> tabToName = Map.of("orchestrator: opus", "opus", "captain: sol", "sol");
Map<String, String> leads = scanner(herdr, tabToName, new AtomicLong()).get();
assertEquals(Map.of("term_opus", "opus", "term_sol", "sol"), leads,
"no shared prefix needed — each lead is matched by its own configured tab");
void anUnlabelledTabContributesNothing() {
assertFalse(scanner(twoLeads(), Map.of(), new AtomicLong()).get().containsKey("term_notes"));
}
/**
@@ -174,28 +148,25 @@ class LeadTabScannerTest {
void aTabInAWorkerSpaceIsNeverALeadEvenWhenItsLabelMatches() {
TopologyHerdr herdr = twoLeads().tab("w9:t2", "w9", "lead: impostor")
.pane("w9:p2", "w9:t2", "term_impostor");
Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
tabToName.put("lead: impostor", "impostor");
assertFalse(scanner(herdr, tabToName, new AtomicLong()).get().containsKey("term_impostor"));
assertFalse(scanner(herdr, Map.of(), new AtomicLong()).get().containsKey("term_impostor"));
}
@Test
void aLabelWithNoConfiguredEntryIsIgnored() {
void aBarePrefixNamesNobodyAndIsRejected() {
TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
.tab("w1:t1", "w1", "lead: nobody-configured").pane("w1:p1", "w1:t1", "term_a");
.tab("w1:t1", "w1", "lead:").pane("w1:p1", "w1:t1", "term_a");
assertEquals(Map.of(), scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get(),
"a label that names no configured lead resolves nobody");
assertEquals(Map.of(), scanner(herdr, Map.of(), new AtomicLong()).get(),
"a lead with no name would resolve as PRIMARY with nothing to attribute it to");
}
@Test
void matchingIsCaseInsensitiveAndToleratesSurroundingWhitespace() {
void thePrefixMatchesCaseInsensitivelyAndTheNameIsTrimmed() {
TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
.tab("w1:t1", "w1", " LEAD: Opus-5.0 ").pane("w1:p1", "w1:t1", "term_a");
.tab("w1:t1", "w1", " LEAD: opus-5.0 ").pane("w1:p1", "w1:t1", "term_a");
assertEquals(Map.of("term_a", "opus-5.0"),
scanner(herdr, Map.of("lead: Opus-5.0", "opus-5.0"), new AtomicLong()).get());
assertEquals(Map.of("term_a", "opus-5.0"), scanner(herdr, Map.of(), new AtomicLong()).get());
}
@Test
@@ -204,51 +175,18 @@ class LeadTabScannerTest {
// nothing bridged placed can land here (see the worker-space test above).
TopologyHerdr herdr = twoLeads().pane("w1:p1b", "w1:t1", "term_opus_split");
assertEquals("opus-5.0",
scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get().get("term_opus_split"));
assertEquals("opus-5.0", scanner(herdr, Map.of(), new AtomicLong()).get().get("term_opus_split"));
}
/**
* CB-579 acceptance (6): this is the bug the ticket closes. A stale pin used to be merged back
* over every scan and never expire; now a scan is the whole answer, so a lead whose tab is gone
* drops out on the very next scan.
*/
@Test
void aTabNoLongerPresentDropsTheLeadOnTheNextScan() {
TopologyHerdr herdr = twoLeads();
AtomicLong clock = new AtomicLong();
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
assertTrue(s.get().containsKey("term_opus"));
void anExplicitlyConfiguredLeadIsMergedInAndOutranksALabel() {
Map<String, String> configured = Map.of("term_opus", "pinned-name", "term_extra", "from-config");
// The session behind term_opus restarted — herdr no longer reports that tab or pane at all.
herdr.tabs.remove("w1:t1");
herdr.panes.remove("w1:p1");
clock.addAndGet(TTL);
Map<String, String> leads = scanner(twoLeads(), configured, new AtomicLong()).get();
assertFalse(s.get().containsKey("term_opus"),
"a stale entry must expire once the tab it named is gone, not be merged back forever");
}
/**
* CB-579 acceptance (5): the whole point of matching by tab instead of {@code terminal_id} — a
* restart changes the terminal, not the tab, so the lead resolves under the same name with no
* config edit.
*/
@Test
void aLeadRestartingInTheSameTabResolvesUnderTheSameName() {
TopologyHerdr herdr = twoLeads();
AtomicLong clock = new AtomicLong();
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
assertEquals("opus-5.0", s.get().get("term_opus"));
// The session restarts: herdr assigns the pane a new terminal_id, same tab (w1:t1).
herdr.panes.remove("w1:p1");
herdr.pane("w1:p1", "w1:t1", "term_opus_v2");
clock.addAndGet(TTL);
Map<String, String> leads = s.get();
assertEquals("opus-5.0", leads.get("term_opus_v2"), "the new terminal resolves immediately");
assertFalse(leads.containsKey("term_opus"), "the old terminal_id is simply gone, not carried");
assertEquals("pinned-name", leads.get("term_opus"), "an explicit pin is the operator's last word");
assertEquals("from-config", leads.get("term_extra"), "a configured lead needs no tab at all");
assertEquals("gpt-sol-5.6", leads.get("term_gpt"));
}
// ── caching ─────────────────────────────────────────────────────────────────────────────────
@@ -257,7 +195,7 @@ class LeadTabScannerTest {
void aSecondLookupWithinTheTtlDoesNotTouchHerdr() {
TopologyHerdr herdr = twoLeads();
AtomicLong clock = new AtomicLong();
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
LeadTabScanner s = scanner(herdr, Map.of(), clock);
s.get();
int afterFirst = herdr.calls;
@@ -272,23 +210,21 @@ class LeadTabScannerTest {
void aTabLabelledAfterStartupIsPickedUpOnceTheTtlExpires() {
TopologyHerdr herdr = twoLeads();
AtomicLong clock = new AtomicLong();
Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
tabToName.put("lead: late-arrival", "late-arrival");
LeadTabScanner s = scanner(herdr, tabToName, clock);
LeadTabScanner s = scanner(herdr, Map.of(), clock);
assertFalse(s.get().containsKey("term_notes"));
herdr.tab("w1:t3", "w1", "lead: late-arrival"); // the operator renames their tab
clock.addAndGet(TTL);
assertEquals("late-arrival", s.get().get("term_notes"),
"the whole point over a config-held terminal_id: no config edit, no restart");
"the whole point over `leaders:`: no config edit, no restart");
}
@Test
void aFailedScanKeepsTheLeadsAlreadyKnownRatherThanDemotingThem() {
TopologyHerdr herdr = twoLeads();
AtomicLong clock = new AtomicLong();
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
LeadTabScanner s = scanner(herdr, Map.of(), clock);
Map<String, String> before = s.get();
herdr.failing = true;
@@ -299,14 +235,14 @@ class LeadTabScannerTest {
}
@Test
void aFailedFirstScanReturnsEmptyRatherThanThrowing() {
void aFailedFirstScanStillHonoursTheConfiguredLeads() {
TopologyHerdr herdr = twoLeads();
herdr.failing = true;
Map<String, String> leads = scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get();
Map<String, String> leads = scanner(herdr, Map.of("term_x", "opus-5.0"), new AtomicLong()).get();
assertEquals(Map.of(), leads,
"with nothing scanned yet and no override to fall back on, the map is simply empty");
assertEquals(Map.of("term_x", "opus-5.0"), leads,
"config-named leads must not depend on herdr answering at all");
}
@Test
@@ -314,7 +250,7 @@ class LeadTabScannerTest {
TopologyHerdr herdr = twoLeads();
herdr.failing = true;
AtomicLong clock = new AtomicLong();
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
LeadTabScanner s = scanner(herdr, Map.of(), clock);
s.get();
int afterFirst = herdr.calls;
@@ -7,14 +7,9 @@ import ch.qos.logback.core.read.ListAppender;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.msg.Rendezvous;
import dev.ltms.bridged.msg.TestTurnTokens;
import dev.ltms.bridged.msg.TurnToken;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.Set;
import java.util.regex.Pattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -26,7 +21,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);
resolver.resolve("term_a", null); // no in-flight turn captured for this target
@@ -38,7 +33,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);
resolver.fail("term_a", null); // no in-flight turn, and no registered waiter to fall back to
@@ -50,9 +45,9 @@ 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);
resolver.captureBaseline("term_a", TestTurnTokens.inert("term_a")); // no send to attribute a later completion to
resolver.captureBaseline("term_a"); // no send to attribute a later completion to
assertFalse(herdr.called("agent.read"),
"with no waiting send there is no turn to baseline — skip the scrape");
@@ -139,7 +134,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);
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 +149,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);
var waiter = rendezvous.open("term_a");
// Delivery baseline was the previous turn's "391"; the scrape now differs → resolve.
@@ -171,7 +166,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);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -185,7 +180,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);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -197,9 +192,9 @@ 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);
var waiter = rendezvous.open("term_a");
resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
resolver.captureBaseline("term_a");
herdr.readText("⏺ answer that /clear would erase\n❯ ");
resolver.resolveBeforePostAction("term_a");
@@ -219,10 +214,10 @@ 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);
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
resolver.captureBaseline("term_a"); // baseline is the clipped >cap block
var turn = resolver.inFlight("term_a");
assertEquals(CompletionResolver.MAX_SCRAPE_CHARS, turn.baseline().length(),
"the delivery baseline is clipped to the same cap resolve() applies to the tail");
@@ -239,7 +234,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);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
@@ -257,7 +252,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);
var waiter = rendezvous.open("term_a");
var turn = new CompletionResolver.InFlight(waiter, ""); // empty pane baselined at delivery
@@ -277,7 +272,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);
var waiter = rendezvous.open("term_a");
var turn = new CompletionResolver.InFlight(waiter, null);
@@ -298,7 +293,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);
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 +319,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);
var waiter = rendezvous.open("term_a");
resolver.fail("term_a", null);
@@ -352,7 +347,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);
var waiterN = rendezvous.open("term_a"); // turn N's send
// The turn as the injector captured it at delivery (waiter + pre-turn baseline).
@@ -373,88 +368,4 @@ class CompletionResolverTest {
"turn N stays resolved by its own reply");
assertTrue(rendezvous.isWaiting("term_a"), "turn N+1 is still awaiting its own resolution");
}
// --- CB-578 stage A: backend-exhausted classification ---------------------------------
@Test
void classifiesAMatchingScrapeAsBackendExhaustedInsteadOfACompletedReply() {
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");
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertTrue(waiter.isDone(), "a matching scrape still resolves the blocked send");
assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
"not reported as a completed reply — the classification is distinct");
}
@Test
void theExhaustedReasonCarriesTheMatchedLine() {
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");
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertEquals("backend exhausted (usage limit): The usage limit has been reached. Try again later.",
waiter.getNow(null).text(), "the reason names the real cause and carries the matched line");
}
@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);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
"a scrape that does not match the pattern is an ordinary completion");
assertEquals("complete report", waiter.getNow(null).text());
}
@Test
void aProfileWithNoConfiguredPatternKeepsTodaysCompletionFallbackUnchanged() {
// Even a scrape that WOULD have matched some other profile's pattern must resolve as a
// plain completion when this target's own profile has none configured (CB-578 criterion 4).
String block = "⏺ The usage limit has been reached.\n❯ ";
FakeHerdr herdr = new FakeHerdr().readText(block);
Rendezvous rendezvous = new Rendezvous();
CompletionResolver resolver =
new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none());
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
"no pattern configured for this target's profile ⇒ unchanged completion-fallback behaviour");
assertEquals("The usage limit has been reached.", waiter.getNow(null).text());
}
@Test
void coverageIsOffWhenNoProfileHasAPatternConfigured() {
assertEquals("off (no profile has an exhaustedPattern configured; profiles: [terra])",
CompletionResolver.coverage(Set.of("terra"), Set.of()));
}
@Test
void coverageIsFullWhenEveryProfileHasAPatternConfigured() {
assertEquals("full (all profiles configured: [gx10, terra])",
CompletionResolver.coverage(Set.of("terra", "gx10"), Set.of("terra", "gx10")));
}
@Test
void coverageIsPartialAndNamesWhichProfilesAreConfigured() {
assertEquals("partial (configured: [terra]; not configured: [gx10])",
CompletionResolver.coverage(Set.of("terra", "gx10"), Set.of("terra")));
}
}
@@ -8,7 +8,6 @@ import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.AgentStatus;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.herdr.HerdrException;
import dev.ltms.bridged.msg.TestTurnTokens;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
@@ -48,7 +47,7 @@ class InjectorTest {
@Test
void deliversWhenIdle() {
CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
CompletableFuture<Void> f = injector.enqueue(T, "hello");
assertFalse(f.isDone(), "not delivered until an injectable status arrives");
injector.onStatus(T, AgentStatus.IDLE);
assertTrue(f.isDone());
@@ -60,7 +59,7 @@ class InjectorTest {
// CB-113: idle alone is not enough — hold until the worker's MCP is connected (ready).
java.util.Set<String> ready = new java.util.HashSet<>();
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // idle but not yet available → held out of the boot window
assertEquals(List.of(), sent(), "must not deliver into a not-yet-available worker");
@@ -80,7 +79,7 @@ class InjectorTest {
void resubmitsEnterWhenADeliveredMessageIsNotPickedUp() {
// CB-113: the Enter at delivery can race the paste; while the worker stays idle (not picked
// up), the injector re-nudges Enter so the pending paste submits.
injector.enqueue(T, "task", TestTurnTokens.inert(T));
injector.enqueue(T, "task");
injector.onStatus(T, AgentStatus.IDLE); // deliver: paste + one Enter
long afterDeliver = enterKeystrokes();
@@ -96,7 +95,7 @@ class InjectorTest {
@Test
void holdsWhileWorkingThenDeliversOnIdle() {
injector.enqueue(T, "later", TestTurnTokens.inert(T));
injector.enqueue(T, "later");
injector.onStatus(T, AgentStatus.WORKING);
assertEquals(List.of(), sent(), "must not inject mid-turn");
injector.onStatus(T, AgentStatus.IDLE);
@@ -105,7 +104,7 @@ class InjectorTest {
@Test
void blockedIsInjectableButUnknownIsNot() {
injector.enqueue(T, "answer", TestTurnTokens.inert(T));
injector.enqueue(T, "answer");
injector.onStatus(T, AgentStatus.UNKNOWN);
assertEquals(List.of(), sent(), "unknown status is not safe to inject");
injector.onStatus(T, AgentStatus.BLOCKED);
@@ -114,8 +113,8 @@ class InjectorTest {
@Test
void twoRapidDeliveriesNeverInterleave() {
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
injector.enqueue(T, "m1");
injector.enqueue(T, "m2");
// First idle window delivers only m1, even if idle is observed twice before pickup.
injector.onStatus(T, AgentStatus.IDLE);
@@ -130,8 +129,8 @@ class InjectorTest {
@Test
void transientUnknownDoesNotReleaseThePickupLatch() {
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
injector.enqueue(T, "m1");
injector.enqueue(T, "m2");
injector.onStatus(T, AgentStatus.IDLE); // m1 sent, awaiting pickup
assertEquals(List.of("m1"), sent());
@@ -146,8 +145,8 @@ class InjectorTest {
@Test
void missedPickupEdgeIsReleasedByGraceSoTheQueueNeverWedges() {
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
injector.enqueue(T, "m1");
injector.enqueue(T, "m2");
injector.onStatus(T, AgentStatus.IDLE); // m1 sent
assertEquals(List.of("m1"), sent());
@@ -159,9 +158,9 @@ class InjectorTest {
@Test
void fifoOrderAcrossManyTurns() {
injector.enqueue(T, "a", TestTurnTokens.inert(T));
injector.enqueue(T, "b", TestTurnTokens.inert(T));
injector.enqueue(T, "c", TestTurnTokens.inert(T));
injector.enqueue(T, "a");
injector.enqueue(T, "b");
injector.enqueue(T, "c");
for (int i = 0; i < 3; i++) {
injector.onStatus(T, AgentStatus.IDLE); // deliver one
injector.onStatus(T, AgentStatus.WORKING); // pickup
@@ -173,7 +172,7 @@ class InjectorTest {
@Test
void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
assertTrue(injector.activeTargets().isEmpty());
injector.enqueue(T, "x", TestTurnTokens.inert(T));
injector.enqueue(T, "x");
assertEquals(Set.of(T), injector.activeTargets(), "active while a message is queued");
injector.onStatus(T, AgentStatus.IDLE); // delivers; awaiting pickup
@@ -192,7 +191,7 @@ class InjectorTest {
void firesTurnCompleteOnAConfirmedWorkingThenIdle() {
List<String> completed = new ArrayList<>();
Injector inj = new Injector(new AgentControl(herdr), completed::add);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // deliver
inj.onStatus(T, AgentStatus.WORKING); // pickup + turn running
@@ -227,8 +226,8 @@ class InjectorTest {
}
ResetListener listener = new ResetListener();
Injector inj = new Injector(agents, listener);
inj.enqueue(T, "first", TestTurnTokens.inert(T));
inj.enqueue(T, "second", TestTurnTokens.inert(T));
inj.enqueue(T, "first");
inj.enqueue(T, "second");
inj.onStatus(T, AgentStatus.IDLE); // first delegation
inj.onStatus(T, AgentStatus.WORKING);
@@ -247,7 +246,7 @@ class InjectorTest {
void doesNotSynthesizeCompletionFromAnUnconfirmedTurn() {
List<String> completed = new ArrayList<>();
Injector inj = new Injector(new AgentControl(herdr), completed::add);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
// Deliver, then only ever idle — a `working` sample is never seen. The pickup grace unwedges
// the queue but must NOT invent a completion: without a sampled turn there is no trustworthy
@@ -260,7 +259,6 @@ class InjectorTest {
private static final class Captor implements TurnListener {
final List<String> completed = new ArrayList<>();
final List<String> failed = new ArrayList<>();
final List<String> failureReasons = new ArrayList<>();
@Override
public void onTurnComplete(String target) {
@@ -271,12 +269,6 @@ class InjectorTest {
public void onTurnFailed(String target) {
failed.add(target);
}
@Override
public void onTurnFailed(String target, String reason) {
failed.add(target);
failureReasons.add(reason);
}
}
// ~30s of unknown at the 250ms prod poll interval; enough onStatus samples to trip the stall.
@@ -286,7 +278,7 @@ class InjectorTest {
void failsAnOutstandingDelegationWhoseWorkerWedgesInUnknown() {
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // deliver
inj.onStatus(T, AgentStatus.WORKING); // worker starts the turn
@@ -301,7 +293,7 @@ class InjectorTest {
void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // deliver
inj.onStatus(T, AgentStatus.WORKING); // confirmed turn
@@ -316,7 +308,7 @@ class InjectorTest {
void sendFailureDropsMessageAndFailsItsFuture() {
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
Injector inj = new Injector(new AgentControl(failing));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "boom");
inj.onStatus(T, AgentStatus.IDLE);
assertTrue(f.isCompletedExceptionally());
@@ -325,35 +317,18 @@ class InjectorTest {
@Test
void dropFailsPendingWaiters() {
CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
CompletableFuture<Void> f = injector.enqueue(T, "orphan");
injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
}
@Test
void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));
inj.onStatus(T, AgentStatus.IDLE); // deliver the first message
inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
inj.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
assertEquals(List.of(T), cap.failed, "drop signals one turn failure for both affected states");
assertEquals(List.of("agent target sol not found"), cap.failureReasons);
assertTrue(delivered.isDone(), "the delivered future has already completed");
assertTrue(queued.isCompletedExceptionally(), "the queued future fails with the drop cause");
}
@Test
void dropFailsTheTurnOfADeliveredMessageWhenTheWorkerVanishes() {
// CB-110: the message was delivered (no longer queued), so failing queued waiters alone would
// leave its send hanging. A vanished worker must fail that in-flight turn too.
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // deliver
inj.onStatus(T, AgentStatus.WORKING); // turn running
@@ -368,7 +343,7 @@ class InjectorTest {
// by an off-sub worker's review of CB-110, delegated through the bridge.)
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
inj.onStatus(T, AgentStatus.IDLE); // deliver; pickup never confirmed
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
@@ -387,7 +362,7 @@ class InjectorTest {
Captor cap = new Captor();
List<String> forgotten = new ArrayList<>();
Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "task");
for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
@@ -415,7 +390,7 @@ class InjectorTest {
try {
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> false, _ -> {
});
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
@@ -440,7 +415,7 @@ class InjectorTest {
Set<String> ready = new java.util.HashSet<>();
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains, _ -> {
});
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.enqueue(T, "task");
for (int i = 0; i < 100; i++) inj.onStatus(T, AgentStatus.IDLE); // still booting, well under grace
assertEquals(List.of(), sent());
@@ -456,7 +431,7 @@ class InjectorTest {
// linger past the worker's life (MemberPresence.forget had no caller before this).
List<String> forgotten = new ArrayList<>();
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, forgotten::add);
inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
inj.enqueue(T, "orphan");
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
assertEquals(List.of(T), forgotten, "drop clears the gone worker's presence");
}
@@ -477,7 +452,7 @@ class InjectorTest {
try {
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, _ -> {
});
inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
inj.enqueue(T, "orphan");
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
String warn = appender.list.stream()
@@ -501,7 +476,7 @@ class InjectorTest {
StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
poller.start();
try {
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller");
delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
} finally {
poller.stop();
@@ -520,7 +495,7 @@ class InjectorTest {
void deliveredFutureCarriesSendFailure() {
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
Injector inj = new Injector(new AgentControl(failing));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "boom");
inj.onStatus(T, AgentStatus.IDLE);
ExecutionException ex = assertThrows(ExecutionException.class, f::get);
assertInstanceOf(HerdrException.class, ex.getCause());
@@ -28,7 +28,7 @@ class LeadLauncherTest {
List.of("ccs", "ltms"), "tab", "bridged-workers", null,
"http://127.0.0.1:8765/mcp", null, null,
null, null, null,
Map.of("CLAUDE_CODE_AUTO_COMPACT_WINDOW", "300000"), null, null, true, null);
Map.of("CLAUDE_CODE_AUTO_COMPACT_WINDOW", "300000"), null, null, true);
}
private static BridgedConfig configWith(BridgedConfig.Leader lead) {
@@ -41,8 +41,8 @@ class LeadLauncherTest {
null, null, fleet, null, "fixed", null).withDefaults();
}
private static BridgedConfig.Leader lead(String profile, String tab, int instances) {
return new BridgedConfig.Leader(profile, tab, instances, "lead:", 10, null, null,
private static BridgedConfig.Leader lead(String profile, String terminal, int instances) {
return new BridgedConfig.Leader(profile, terminal, instances, "lead:", 10, null, null,
"leads", "/repo");
}
@@ -70,16 +70,16 @@ class LeadLauncherTest {
void startsTheDeclaredLeadWhenNoneIsRunning() {
FakeHerdr herdr = new FakeHerdr();
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
assertTrue(herdr.called("agent.start"), "a lead must actually be started");
assertEquals("lead-opus", startedName(herdr));
}
/** The tab is labelled with the configured `tab:` so the scanner finds the lead on the next resolve. */
/** The tab is labelled so the scanner finds the lead on the next resolve. */
@Test
void labelsTheTabWithTheConfiguredTabValue() {
void labelsTheTabWithThePrefixTheScannerReadsBack() {
FakeHerdr herdr = new FakeHerdr();
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
assertEquals("lead: opus",
((Map<?, ?>) herdr.lastCall("tab.rename").params()).get("label"));
@@ -90,7 +90,7 @@ class LeadLauncherTest {
void startsAsManyInstancesAsAreDeclared() {
FakeHerdr herdr = new FakeHerdr();
assertEquals(2, launcher(herdr, configWith(lead("opus", "lead: opus", 2))).ensureLeads());
assertEquals(2, launcher(herdr, configWith(lead("opus", null, 2))).ensureLeads());
assertEquals(2, herdr.calls.stream().filter(c -> c.method().equals("agent.start")).count());
}
@@ -104,7 +104,7 @@ class LeadLauncherTest {
.withTab("wL", "wL:t1", "lead: opus")
.withAgent("lead-opus", "term_lead", "wL:p1", "wL:t1");
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
assertFalse(herdr.called("agent.start"), "the live lead must not be duplicated");
}
@@ -118,23 +118,20 @@ class LeadLauncherTest {
.withWorkspace("wL", "leads")
.withTab("wL", "wL:t1", "lead: opus"); // label only — nothing running in it
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
"a stale label is not a lead; the lead must be relaunched");
}
/**
* A lead the operator opened by hand is live once its tab carries the configured `tab:` label —
* CB-579 retired the `terminal:` pin, so a hand-opened lead is found the same way an
* auto-launched one is, by its tab, not by a terminal id nobody wrote down in advance.
* A lead the operator opened by hand and pinned with `terminal:` is live even though its tab
* carries no matching label. Counting labels alone would relaunch it on every boot.
*/
@Test
void aHandOpenedLeadWithTheConfiguredTabLabelCountsAsLive() {
void aPinnedTerminalWithARunningAgentCountsAsLive() {
FakeHerdr herdr = new FakeHerdr()
.withWorkspace("wX", "main")
.withTab("wX", "wX:t1", "lead: opus")
.withAgent("hand-opened", "term_hand", "wX:p1", "wX:t1");
.withAgent("hand-opened", "term_pinned", "wX:p1", "wX:t1");
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead("opus", "term_pinned", 1))).ensureLeads());
assertFalse(herdr.called("agent.start"));
}
@@ -146,7 +143,7 @@ class LeadLauncherTest {
.withTab("wM", "wM:t1", "lead: opus") // a member tab that looks like a lead
.withAgent("claude-opus-x", "term_m", "wM:p1", "wM:t1");
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
"a member in a lead-labelled tab is not a lead, so the real lead is still missing");
}
@@ -155,7 +152,7 @@ class LeadLauncherTest {
void anUncountableHerdrStartsNothing() {
FakeHerdr herdr = new FakeHerdr().healthy(false);
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
assertFalse(herdr.called("agent.start"));
}
@@ -169,7 +166,7 @@ class LeadLauncherTest {
@Test
void theLeadNeverReceivesTheWorkerReplyCharter() {
FakeHerdr herdr = new FakeHerdr();
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
List<String> args = startedArgs(herdr);
assertFalse(args.contains("--append-system-prompt"),
@@ -181,7 +178,7 @@ class LeadLauncherTest {
@Test
void theLeadMountsTheBridgeMcpAndPinsItsModel() {
FakeHerdr herdr = new FakeHerdr();
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
List<String> args = startedArgs(herdr);
assertTrue(args.contains("--mcp-config"));
@@ -195,7 +192,7 @@ class LeadLauncherTest {
@Test
void theLeadEnvCarriesNoAnthropicBinding() {
FakeHerdr herdr = new FakeHerdr();
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
Map<String, String> env = tabEnv(herdr);
assertNull(env.get("ANTHROPIC_BASE_URL"));
@@ -208,7 +205,7 @@ class LeadLauncherTest {
@Test
void theLeadTabIsCreatedOutsideEveryMemberWorkspace() {
FakeHerdr herdr = new FakeHerdr();
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
String label = (String) ((Map<?, ?>) herdr.lastCall("workspace.create").params()).get("label");
assertEquals("leads", label);
@@ -217,12 +214,12 @@ class LeadLauncherTest {
// ── recognise-only and misconfiguration ───────────────────────────────────────────────────
/** A lead with a tab but no profile is recognise-only by design — not an error, not a launch. */
/** A lead with a pin but no profile is recognise-only by design — not an error, not a launch. */
@Test
void aLeadThatNamesNoProfileIsRecognisedButNeverLaunched() {
FakeHerdr herdr = new FakeHerdr();
assertEquals(0, launcher(herdr, configWith(lead(null, "lead: dead", 1))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead(null, "term_dead", 1))).ensureLeads());
assertFalse(herdr.called("agent.start"));
}
@@ -231,7 +228,7 @@ class LeadLauncherTest {
void zeroInstancesLaunchesNothing() {
FakeHerdr herdr = new FakeHerdr();
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 0))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 0))).ensureLeads());
assertFalse(herdr.called("agent.start"));
}
@@ -240,7 +237,7 @@ class LeadLauncherTest {
void anUnknownProfileIsSkippedRatherThanThrown() {
FakeHerdr herdr = new FakeHerdr();
assertEquals(0, launcher(herdr, configWith(lead("nope", "lead: opus", 1))).ensureLeads());
assertEquals(0, launcher(herdr, configWith(lead("nope", null, 1))).ensureLeads());
assertFalse(herdr.called("agent.start"));
}
@@ -1,37 +0,0 @@
package dev.ltms.bridged.logging;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import io.modelcontextprotocol.spec.McpSchema;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
class McpCancelledNotificationFilterTest {
@Test
void suppressesOnlyTheCancelledNotificationWarning() {
Logger logger = (Logger) LoggerFactory.getLogger(McpCancelledNotificationFilter.LOGGER);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
try {
logger.warn(McpCancelledNotificationFilter.UNHANDLED_NOTIFICATION,
new McpSchema.JSONRPCNotification("notifications/cancelled", Map.of("requestId", 7)));
logger.warn(McpCancelledNotificationFilter.UNHANDLED_NOTIFICATION,
new McpSchema.JSONRPCNotification("notifications/progress", Map.of("progress", 1)));
assertEquals(1, appender.list.size());
assertEquals(Level.WARN, appender.list.getFirst().getLevel());
assertTrue(appender.list.getFirst().getFormattedMessage().contains("notifications/progress"));
} finally {
logger.detachAppender(appender);
}
}
}
@@ -72,7 +72,7 @@ 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);
return mcp;
}
@@ -53,18 +53,6 @@ class BridgeMcpTest {
return ((McpSchema.TextContent) r.content().getFirst()).text();
}
private void assertSendRoundTrips(String target, Set<String> profiles) throws Exception {
CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
() -> BridgeMcp.send(messages, target, "hi", 4000L, null, profiles));
long deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting(target) && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
assertTrue(rendezvous.isWaiting(target), "send should be accepted for " + target);
BridgeMcp.reply(messages, target, "received");
assertEquals("received", textOf(send.get(6, TimeUnit.SECONDS)));
}
private static ClaudeCodeLauncher workerService(FakeHerdr h, String baseUrl, Set<String> allow) {
BridgedConfig.Profile cfg = new BridgedConfig.Profile(
"ltms-local", baseUrl, "coder", null, "BRIDGED_WORKER_TOKEN", null,
@@ -81,7 +69,7 @@ class BridgeMcpTest {
void sendThenReplyRoundTrips() throws Exception {
// bridge_send blocks; bridge_reply resolves it with the worker's structured answer.
CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
() -> BridgeMcp.send(messages, "term_a", "review this", 4000L, null, Set.of()));
() -> BridgeMcp.send(messages, "term_a", "review this", 4000L));
// Wait until the send has opened its waiter so the reply resolves it (CB-307: reply now
// queues in the inbox if no waiter is open, which would break the round-trip).
@@ -103,7 +91,7 @@ class BridgeMcpTest {
@Test
void asyncSendReturnsATicketThenPollReportsTheReply() throws Exception {
// wait:false parity — a ticket is issued, resolved by a reply, and surfaced by bridge_poll.
McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it");
assertNotEquals(Boolean.TRUE, accepted.isError());
String out = textOf(accepted);
assertTrue(out.contains("ticket="), out);
@@ -132,131 +120,6 @@ class BridgeMcpTest {
assertEquals("async LGTM", textOf(polled));
}
@Test
void asyncSendSurfacesAnAskThenKeepsTheTicketForTheFinalReply() throws Exception {
McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
long deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
assertTrue(rendezvous.isWaiting("term_a"));
CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
() -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
McpSchema.CallToolResult question = BridgeMcp.poll(messages, ticket, null);
deadline = System.currentTimeMillis() + 3000;
while (!textOf(question).contains("[question") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
question = BridgeMcp.poll(messages, ticket, null);
}
assertTrue(textOf(question).contains("which config?"), textOf(question));
String questionText = textOf(question);
String afterTurnId = questionText.substring(questionText.indexOf("turnId=\"") + "turnId=\"".length());
String turnId = afterTurnId.substring(0, afterTurnId.indexOf('"'));
CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
() -> BridgeMcp.answer(messages, turnId, "config.yaml", 5000L));
assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
assertTrue(rendezvous.isWaiting("term_a"));
BridgeMcp.reply(messages, "term_a", "done");
assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
McpSchema.CallToolResult done = BridgeMcp.poll(messages, ticket, null);
deadline = System.currentTimeMillis() + 3000;
while (!"done".equals(textOf(done)) && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
done = BridgeMcp.poll(messages, ticket, null);
}
assertEquals("done", textOf(done));
}
@Test
void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
long deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
assertTrue(rendezvous.isWaiting("term_a"));
McpSchema.CallToolResult ask = BridgeMcp.ask(messages, "term_a", "still there?", 50L);
assertTrue(textOf(ask).contains("no answer"), textOf(ask));
assertTrue(textOf(BridgeMcp.poll(messages, ticket, null)).startsWith("[pending"));
BridgeMcp.reply(messages, "term_a", "finished after timeout");
assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
}
@Test
void asyncSendFailureDoesNotLeaveItsTicketPending() throws Exception {
String ticket = messages.sendAsync("term_a", "do it", () -> {
throw new IllegalStateException("accept failed");
});
long deadline = System.currentTimeMillis() + 3000;
MessageService.TaskView view = messages.poll(ticket);
while (view.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
view = messages.poll(ticket);
}
assertEquals(MessageService.Phase.FAILED, view.phase());
assertEquals("accept failed", view.detail());
}
@Test
void anotherAsyncTicketCannotCaptureAReplyWhileTheFirstTicketIsAsking() throws Exception {
McpSchema.CallToolResult firstAccepted = BridgeMcp.sendAsync(messages, "term_a", "first", null, Set.of());
String firstTicket = textOf(firstAccepted).substring(textOf(firstAccepted).indexOf("ticket=") + "ticket=".length()).trim();
long deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
assertTrue(rendezvous.isWaiting("term_a"));
CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
() -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
MessageService.TaskView first = messages.poll(firstTicket);
deadline = System.currentTimeMillis() + 3000;
while (first.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
first = messages.poll(firstTicket);
}
assertEquals(MessageService.Phase.ASKING, first.phase());
String firstTurnId = first.turnId();
McpSchema.CallToolResult secondAccepted = BridgeMcp.sendAsync(messages, "term_a", "second", null, Set.of());
String secondTicket = textOf(secondAccepted).substring(textOf(secondAccepted).indexOf("ticket=") + "ticket=".length()).trim();
MessageService.TaskView second = messages.poll(secondTicket);
deadline = System.currentTimeMillis() + 3000;
while (second.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
second = messages.poll(secondTicket);
}
assertEquals(MessageService.Phase.FAILED, second.phase());
BridgeMcp.reply(messages, "term_a", "late reply");
assertEquals("late reply", messages.drainReplies("term_a").getFirst().content());
CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
() -> BridgeMcp.answer(messages, firstTurnId, "config.yaml", 5000L));
assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
BridgeMcp.reply(messages, "term_a", "done");
assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
}
@Test
void pollUnknownTicketIsAnError() {
McpSchema.CallToolResult res = BridgeMcp.poll(messages, "task-999", null);
@@ -266,40 +129,15 @@ class BridgeMcpTest {
@Test
void sendTimesOutWithAWorkingNote() {
McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L, null, Set.of());
McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L);
assertNotEquals(Boolean.TRUE, res.isError(), "a timeout is informational, not a tool error");
assertTrue(textOf(res).contains("no reply"), "got: " + textOf(res));
}
@Test
void sendRejectsMissingArgs() {
assertTrue(BridgeMcp.send(messages, null, "hi", null, null, Set.of()).isError());
assertTrue(BridgeMcp.send(messages, "term_a", " ", null, null, Set.of()).isError());
}
@Test
void sendRejectsAConfiguredProfileNameBeforeAcceptingIt() {
McpSchema.CallToolResult blocking = BridgeMcp.send(messages, "sol", "hi", 100L, null, Set.of("sol"));
McpSchema.CallToolResult async = BridgeMcp.sendAsync(messages, "sol", "hi", null, Set.of("sol"));
assertTrue(blocking.isError());
assertTrue(async.isError());
assertTrue(textOf(blocking).contains("sol"));
assertTrue(textOf(blocking).contains("configured profile name"));
assertTrue(textOf(blocking).contains("bridge_list"));
assertFalse(textOf(async).contains("ticket="));
}
@Test
void sendAllowsPeerLeadMemberAndUnclassifiedTargets() throws Exception {
Set<String> profiles = Set.of("sol");
assertSendRoundTrips("term_peer_lead", profiles);
assertSendRoundTrips("term_live_member", profiles);
// A herdr-owned pane outside the bridge roster cannot be classified at accept time.
McpSchema.CallToolResult result = BridgeMcp.send(messages, "external-pane", "hi", 10L, null, profiles);
assertFalse(result.isError(), "an unclassified target must not be rejected at acceptance time");
assertTrue(BridgeMcp.send(messages, null, "hi", null).isError());
assertTrue(BridgeMcp.send(messages, "term_a", " ", null).isError());
}
@Test
@@ -335,7 +173,7 @@ class BridgeMcpTest {
void askThenAnswerRoundTrips() throws Exception {
// The primary delegates and blocks; wait until its waiter is open before the worker asks.
CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
() -> BridgeMcp.send(messages, "term_a", "do X", 5000L, null, Set.of()));
() -> BridgeMcp.send(messages, "term_a", "do X", 5000L));
long deadline = System.currentTimeMillis() + 3000;
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
//noinspection BusyWait
@@ -467,42 +305,6 @@ class BridgeMcpTest {
assertTrue(out.contains("\"liveStatus\":\"unknown\""), out);
}
@Test
void capacityUsesThePlacementLiveCount() {
FakeHerdr h = new FakeHerdr();
SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
sessions.acquire("ltms-local", null, null, null);
McpSchema.CallToolResult res = BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
sessions, null, new BridgeMcp.CapacitySource(profile -> 2, profile -> 2,
() -> Set.of("ltms-local"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), "");
String out = textOf(res);
assertTrue(out.contains("\"maxLoad\":2"), out);
assertTrue(out.contains("\"live\":2"), out);
assertTrue(out.contains("\"free\":0"), out);
}
@Test
void capacityIncludesConfiguredProfileWithoutMembers() {
FakeHerdr h = new FakeHerdr();
SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
sessions, null, new BridgeMcp.CapacitySource(profile -> 0, profile -> 2,
() -> Set.of("terra"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
assertTrue(out.contains("\"profile\":\"terra\""), out);
assertTrue(out.contains("\"live\":0"), out);
assertTrue(out.contains("\"free\":2"), out);
assertTrue(out.contains("\"reclaimable\":0"), out);
}
@Test
void inertCapacitySourceOmitsCapacityBlock() {
FakeHerdr h = new FakeHerdr();
String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw"))), null,
BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
assertFalse(out.contains("\"capacity\":"), out);
}
@Test
void listReportsLeadsAndFlagsTheCallersOwnRow() {
FakeHerdr h = new FakeHerdr();
@@ -776,7 +578,7 @@ class BridgeMcpTest {
assertTrue(out.contains("\"role\":\"dev\""), out);
assertTrue(out.contains("\"role\":\"reviewer\""), out);
assertEquals(2, out.split("\"profile\":\"ltms-local\"", -1).length - 1,
"inert capacity is omitted, leaving the two member rows: " + out);
"both members share one profile — that is the point: " + out);
}
@Test
@@ -102,53 +102,6 @@ class ClaudeCodeLauncherTest {
"the operator's own args are preserved, in order, ahead of the model flag");
}
@Test
void appendsTheBaseComposedRoleAndReplyCharter() {
FakeHerdr herdr = new FakeHerdr();
String roleCharter = "You review changes.";
BridgedConfig.Profile cfg = new BridgedConfig.Profile(
"sonnet", "http://gx00.gw:8000", null, null, "BRIDGED_WORKER_TOKEN",
List.of("claude"), "tab", "bridged-workers", "w #{n}",
"http://127.0.0.1:8765/mcp", null, null);
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null,
0, 0L, () -> fleet(Map.of("reviewer", roleCharter), null));
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
List<String> args = spawnedArgs(herdr);
int flag = args.indexOf("--append-system-prompt");
assertEquals(1, args.stream().filter("--append-system-prompt"::equals).count(),
"the composed charter is passed once");
assertEquals(roleCharter + "\n\n" + HerdrPeerLauncher.REPLY_CHARTER, args.get(flag + 1),
"the role charter comes first and the reply rule comes last");
}
@Test
void profileWithoutMcpOrRoleCharterGetsNoSystemPrompt() {
FakeHerdr herdr = new FakeHerdr();
ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of(), null));
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.DEV));
assertFalse(spawnedArgs(herdr).contains("--append-system-prompt"));
}
@Test
void profileWithoutMcpStillGetsItsRoleCharter() {
FakeHerdr herdr = new FakeHerdr();
String roleCharter = "You design changes.";
ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of("architect", roleCharter), null));
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.ARCHITECT));
List<String> args = spawnedArgs(herdr);
int flag = args.indexOf("--append-system-prompt");
assertTrue(flag >= 0, "a role charter does not need an MCP mount");
assertEquals(roleCharter, args.get(flag + 1));
assertFalse(args.contains("--mcp-config"));
}
private ClaudeCodeLauncher multiProfile(FakeHerdr herdr) {
BridgedConfig.Profile gx10 = new BridgedConfig.Profile("gx10", "http://gx10.gw:8000", "coder",
null, "BRIDGED_WORKER_TOKEN", List.of("claude"), "tab", "bridged-workers", "w #{n}", null, null, null);
@@ -733,7 +686,7 @@ class ClaudeCodeLauncherTest {
return new BridgedConfig.Profile(
profile, baseUrl, "sonnet", null, "BRIDGED_WORKER_TOKEN",
List.of("ccs", profile), "tab", "bridged-workers", "w #{n}", null, null, null,
null, null, null, Map.of(), null, null, true, null);
null, null, null, Map.of(), null, null, true);
}
@Test
@@ -819,7 +772,7 @@ class ClaudeCodeLauncherTest {
null, null, null,
Map.of("ANTHROPIC_BASE_URL", "http://evil.example.com",
"ANTHROPIC_AUTH_TOKEN", "sk-ant-bad", "JAVA_HOME", "/opt/jdk"),
null, null, true, null);
null, null, true);
new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
_ -> "would-be-token").spawn();
@@ -843,18 +796,14 @@ class ClaudeCodeLauncherTest {
.toList();
}
private static BridgedConfig.Fleet fleet(Map<String, String> charters, String tabLabel) {
return new BridgedConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(), charters, tabLabel);
}
/** A profile with no {@code tabLabel:} of its own — the fleet template decides. */
private ClaudeCodeLauncher labelService(FakeHerdr herdr, Supplier<BridgedConfig.Fleet> fleet) {
private ClaudeCodeLauncher labelService(FakeHerdr herdr, Supplier<String> fleetTemplate) {
BridgedConfig.Profile cfg = new BridgedConfig.Profile(
"sonnet", "http://gx00.gw:8000", "sonnet", null, "BRIDGED_WORKER_TOKEN",
List.of("claude"), "tab", "bridged-workers", null, null, null, null);
return new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
_ -> null, 0, 0L, fleet);
_ -> null, 0, 0L, fleetTemplate);
}
/**
@@ -865,7 +814,7 @@ class ClaudeCodeLauncherTest {
@Test
void theFleetTemplateNamesTheRoleTheMemberWasSpawnedFor() {
FakeHerdr herdr = new FakeHerdr();
ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(null, "{role}: {profile} #{n}"));
ClaudeCodeLauncher svc = labelService(herdr, () -> "{role}: {profile} #{n}");
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
@@ -876,7 +825,7 @@ class ClaudeCodeLauncherTest {
@Test
void theCounterRunsPerRoleAndProfileNotPerFleet() {
FakeHerdr herdr = new FakeHerdr();
ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(null, "{role}: {profile} #{n}"));
ClaudeCodeLauncher svc = labelService(herdr, () -> "{role}: {profile} #{n}");
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.DEV));
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
@@ -890,7 +839,7 @@ class ClaudeCodeLauncherTest {
@Test
void aBlankFleetTemplateFallsBackToTheRoleFirstDefault() {
FakeHerdr herdr = new FakeHerdr();
labelService(herdr, () -> fleet(null, null)).spawn(
labelService(herdr, () -> null).spawn(
new SpawnRequest("sonnet", null, null, null, null, MemberRole.ARCHITECT));
assertEquals(List.of("architect: sonnet #1"), tabLabels(herdr));
@@ -906,7 +855,7 @@ class ClaudeCodeLauncherTest {
List.of("claude"), "tab", "bridged-workers", "pinned {profile}", null, null, null);
new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
_ -> null, 0, 0L, () -> fleet(null, "{role}: {profile} #{n}"))
_ -> null, 0, 0L, () -> "{role}: {profile} #{n}")
.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
assertEquals(List.of("pinned sonnet"), tabLabels(herdr));
@@ -921,7 +870,7 @@ class ClaudeCodeLauncherTest {
void theTemplateIsReadOnEverySpawnSoAnEditTakesEffect() {
FakeHerdr herdr = new FakeHerdr();
AtomicReference<String> template = new AtomicReference<>("{role}: {profile} #{n}");
ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(null, template.get()));
ClaudeCodeLauncher svc = labelService(herdr, template::get);
svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.DEV));
template.set("[{profile}] {role} {n}");
@@ -77,7 +77,7 @@ class CompositePeerLauncherTest {
}
@Override
protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {
protected Launch buildLaunch(BridgedConfig.Profile cfg) {
return new Launch(Map.of(), List.of());
}
@@ -1,71 +0,0 @@
package dev.ltms.bridged.member;
import dev.ltms.bridged.config.BridgedConfig;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.herdr.WorkspaceControl;
import dev.ltms.bridged.peer.Capability;
import dev.ltms.bridged.peer.MemberRole;
import dev.ltms.bridged.peer.SpawnRequest;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
class HerdrPeerLauncherCharterTest {
@Test
void readsAndComposesTheFleetCharterForEachSpawn() {
AtomicReference<BridgedConfig.Fleet> fleet = new AtomicReference<>(fleet(Map.of()));
CapturingLauncher launcher = new CapturingLauncher(fleet::get);
launcher.spawn(new SpawnRequest("mcp", null, null, null, null, MemberRole.DEV));
fleet.set(fleet(Map.of("dev", "role charter")));
launcher.spawn(new SpawnRequest("mcp", null, null, null, null, MemberRole.DEV));
launcher.spawn(new SpawnRequest("no-mcp", null, null, null, null, MemberRole.DEV));
assertEquals(HerdrPeerLauncher.REPLY_CHARTER, launcher.specs.get(0).charter(),
"without a role charter, MCP profiles receive only the reply charter");
assertEquals("role charter\n\n" + HerdrPeerLauncher.REPLY_CHARTER, launcher.specs.get(1).charter(),
"the changed supplier value is read for the next spawn and the reply rule is last");
assertEquals("role charter", launcher.specs.get(2).charter(),
"a role charter does not depend on an MCP mount");
}
private static BridgedConfig.Fleet fleet(Map<String, String> charters) {
return new BridgedConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(), charters, null);
}
private static final class CapturingLauncher extends HerdrPeerLauncher {
private final List<LaunchSpec> specs = new ArrayList<>();
CapturingLauncher(Supplier<BridgedConfig.Fleet> fleet) {
super("test", new AgentControl(new FakeHerdr()), new WorkspaceControl(new FakeHerdr()),
Map.of("mcp", profile("mcp", "http://bridge"),
"no-mcp", profile("no-mcp", null)),
"mcp", _ -> null, 0, () -> 0L, () -> { }, fleet);
}
@Override
protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {
specs.add(spec);
return new Launch(Map.of(), List.of("test"));
}
@Override
public Set<Capability> capabilities() {
return Set.of();
}
private static BridgedConfig.Profile profile(String name, String mcpUrl) {
return new BridgedConfig.Profile(name, "http://gx00.gw:8000", null, null,
"BRIDGED_WORKER_TOKEN", List.of("test"), "pane", null, null, mcpUrl, null, null);
}
}
}
@@ -17,10 +17,6 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.*;
@@ -38,19 +34,12 @@ class OpenCodeLauncherTest {
}
/** Gate-disabled launcher whose per-spawn config dirs land under an inspectable temp root. */
private static OpenCodeLauncher service(FakeHerdr herdr, Path configRoot, BridgedConfig.Profile cfg) {
private OpenCodeLauncher service(FakeHerdr herdr, Path configRoot, BridgedConfig.Profile cfg) {
return new OpenCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
Map.of(cfg.profile(), cfg), cfg.profile(), k -> "GITEA_ACCESS_TOKEN".equals(k) ? "tok" : null,
0, System::currentTimeMillis, () -> { }, configRoot, configRoot);
}
private static OpenCodeLauncher service(FakeHerdr herdr, Path configRoot, BridgedConfig.Profile cfg,
Supplier<BridgedConfig.Fleet> fleet) {
return new OpenCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null,
0, System::currentTimeMillis, () -> { }, configRoot, configRoot, fleet);
}
@SuppressWarnings("unchecked")
private static Map<String, Object> lastStart(FakeHerdr herdr) {
return (Map<String, Object>) herdr.lastCall("agent.start").params();
@@ -73,10 +62,8 @@ class OpenCodeLauncherTest {
@Test
void writesRemoteMcpConfigAndCharterInstructionsWhenMcpUrlSet(@TempDir Path root) throws Exception {
FakeHerdr herdr = new FakeHerdr();
BridgedConfig.Fleet fleet = new BridgedConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(),
Map.of("dev", "role rule"), null);
service(herdr, root, opencodeCfg("google/gemini-2.5-pro", "http://127.0.0.1:8765/mcp", null),
() -> fleet).spawn();
service(herdr, root, opencodeCfg("google/gemini-2.5-pro", "http://127.0.0.1:8765/mcp", null))
.spawn();
Map<String, String> env = startEnv(herdr);
assertNull(env.get("ANTHROPIC_BASE_URL"), "opencode carries no ANTHROPIC_* / subscription boundary");
@@ -95,15 +82,13 @@ class OpenCodeLauncherTest {
"the profile's bridge MCP url is present");
assertTrue(bridge.path("enabled").asBoolean(), "the bridge server is enabled");
assertTrue(json.path("instructions").isArray() && !json.path("instructions").isEmpty(),
"the member charter is mounted via instructions");
"the reply charter is mounted via instructions");
// The instructions entry is a real file path holding the composed member charter.
Path charter = Path.of(cfgPath).resolveSibling("member-charter.md");
// The instructions entry is a real file path holding the reply charter.
Path charter = Path.of(cfgPath).resolveSibling("reply-charter.md");
assertTrue(Files.exists(charter), "the charter file the config references was written");
assertEquals("role rule\n\n" + HerdrPeerLauncher.REPLY_CHARTER, Files.readString(charter),
"the composed charter keeps the role rule first and the reply rule last");
assertEquals(charter.toAbsolutePath().toString(), json.path("instructions").get(0).asText(),
"instructions names the charter file by its absolute path");
assertTrue(Files.readString(charter).contains("bridge_reply"),
"the charter instructs the worker to answer via bridge_reply");
}
@Test
@@ -115,69 +100,6 @@ class OpenCodeLauncherTest {
"no bridge MCP url → no config file and no OPENCODE_CONFIG");
}
@Test
void roleCharterWithoutMcpOrCustomProviderStillWritesAConfig(@TempDir Path root) throws Exception {
FakeHerdr herdr = new FakeHerdr();
BridgedConfig.Fleet fleet = new BridgedConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(),
Map.of("dev", "role rule"), null);
Path configRoot = Files.createDirectory(root.resolve("configs"));
Path checkout = Files.createDirectory(root.resolve("checkout"));
service(herdr, configRoot, opencodeCfg("google/gemini-2.5-pro", null, null), () -> fleet).spawn();
String cfgPath = startEnv(herdr).get("OPENCODE_CONFIG");
assertNotNull(cfgPath, "a role charter needs a config even without MCP or custom provider");
JsonNode json = new ObjectMapper().readTree(Path.of(cfgPath).toFile());
Path charter = Path.of(json.path("instructions").get(0).asText());
assertEquals("role rule", Files.readString(charter), "the base-composed role charter is unchanged");
assertTrue(json.path("mcp").isMissingNode(), "a charter does not add an MCP mount");
assertTrue(charter.startsWith(configRoot), "the charter is written under the temp config root");
try (var files = Files.walk(checkout)) {
assertFalse(files.anyMatch(path -> path.getFileName().toString().equals("member-charter.md")),
"the worker checkout receives no charter file");
}
}
@Test
void nullCharterWritesNoCharterFileOrInstructions(@TempDir Path root) throws Exception {
FakeHerdr herdr = new FakeHerdr();
service(herdr, root, pinnedCfg("local-vllm/model", "http://127.0.0.1:8000", null),
() -> new BridgedConfig.Fleet(Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), null)).spawn();
String config = startEnv(herdr).get("OPENCODE_CONFIG");
assertNotNull(config, "the custom provider still needs a config");
JsonNode json = new ObjectMapper().readTree(Path.of(config).toFile());
assertTrue(json.path("instructions").isMissingNode(), "a null charter adds no instructions entry");
try (var files = Files.walk(root)) {
assertFalse(files.anyMatch(path -> path.getFileName().toString().equals("member-charter.md")),
"a null charter creates no charter file");
}
}
@Test
void concurrentSpawnsWriteSeparateCharterDirectories(@TempDir Path root) throws Exception {
BridgedConfig.Profile cfg = opencodeCfg("google/gemini-2.5-pro", "http://127.0.0.1:8765/mcp", null);
ExecutorService executor = Executors.newFixedThreadPool(2);
try {
Future<String> first = executor.submit(() -> spawnConfigPath(root, cfg));
Future<String> second = executor.submit(() -> spawnConfigPath(root, cfg));
Path firstCharter = Path.of(first.get()).resolveSibling("member-charter.md");
Path secondCharter = Path.of(second.get()).resolveSibling("member-charter.md");
assertNotEquals(firstCharter.getParent(), secondCharter.getParent(),
"each concurrent spawn owns a separate config directory");
assertTrue(Files.exists(firstCharter));
assertTrue(Files.exists(secondCharter));
} finally {
executor.shutdownNow();
}
}
private static String spawnConfigPath(Path root, BridgedConfig.Profile cfg) {
FakeHerdr herdr = new FakeHerdr();
service(herdr, root, cfg).spawn();
return startEnv(herdr).get("OPENCODE_CONFIG");
}
@Test
void passesTheModelAsDashMFlagAlongsideAutoApprove(@TempDir Path root) {
FakeHerdr herdr = new FakeHerdr();
@@ -5,7 +5,6 @@ import dev.ltms.bridged.herdr.AgentStatus;
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.mcp.PrimaryRegistry;
import dev.ltms.bridged.inject.Injector;
import org.junit.jupiter.api.BeforeEach;
@@ -32,8 +31,7 @@ class MessageServiceTest {
private final FakeHerdr herdr = new FakeHerdr().readText("BUILD GREEN: 391 files");
private final AgentControl agents = new AgentControl(herdr);
private final Rendezvous rendezvous = new Rendezvous();
private final CompletionResolver completion =
new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none());
private final CompletionResolver completion = new CompletionResolver(agents, rendezvous);
private final Injector injector = new Injector(agents, completion);
private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
private final MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
@@ -122,36 +120,6 @@ class MessageServiceTest {
assertFalse(reply.completed());
}
@Test
void droppedQueuedAndDeliveredTurnsExposeTheRealCauseExactlyOnce() throws Exception {
CompletableFuture<MessageService.Reply> first = sendAsync();
awaitWaiting();
injector.onStatus(T, AgentStatus.IDLE); // first delivery
injector.onStatus(T, AgentStatus.WORKING); // first turn in flight
CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
MessageService.Reply reply = first.get(5, TimeUnit.SECONDS);
assertEquals(MessageService.Outcome.WORKER_FAILED, reply.outcome());
assertEquals("agent target sol not found", reply.text());
assertTrue(queued.isCompletedExceptionally(), "the queued delivery future also fails");
assertFalse(rendezvous.resolveFailure(waiter, "second failure"), "the waiter fails exactly once");
}
@Test
void noDropReasonKeepsTheExistingFallbackText() throws Exception {
herdr.readText("");
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.open(T);
completion.onTurnFailed(T);
Rendezvous.Resolution resolution = waiter.get(2, TimeUnit.SECONDS);
assertEquals("worker did not reply; its turn ended in an unrecoverable state "
+ "(worker unreachable or stuck)", resolution.text());
}
// --- bridge_ask reverse rendezvous (CB-205) ------------------------------------------------
@Test
@@ -641,94 +609,4 @@ class MessageServiceTest {
assertTrue(view.detail() != null && view.detail().contains("released"),
"and the detail says why, rather than 'worker unknown'");
}
@Test
void abandonFailsEveryPendingAsyncTicketForTheReleasedTarget() throws Exception {
String first = messages.sendAsync(T, "first task");
awaitWaiting(); // first task owns the target lock and rendezvous waiter
String second = messages.sendAsync(T, "second task"); // parked on the same lock, not yet queued
String third = messages.sendAsync(T, "third task"); // a second queued ticket proves the full sweep
assertTrue(messages.abandon(T, "agent target term_a not found"));
assertFailedTicket(first, "agent target term_a not found");
assertFailedTicket(second, "agent target term_a not found");
assertFailedTicket(third, "agent target term_a not found");
}
@Test
void abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
assertFalse(messages.abandon(T, "agent target term_a not found"),
"an asking ticket is an active turn, not a pending send to sweep");
assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase());
CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
() -> messages.answer(asking.turnId(), "config.yaml", 5000));
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
}
@Test
void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
assertEquals(MessageService.AskOutcome.TIMED_OUT,
messages.ask(T, "which config?", 200).outcome());
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
"only the question wait ended; the delegated turn may still finish");
String next = messages.sendAsync(T, "next task");
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
}
@Test
void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
String first = messages.sendAsync(T, "first task");
awaitWaiting();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
assertEquals(MessageService.Phase.ASKING, awaitTicketPhase(first, MessageService.Phase.ASKING).phase());
MessageService.TaskView asking = messages.poll(first);
CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
() -> messages.answer(asking.turnId(), "config.yaml", 5000));
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
}
private void assertFailedTicket(String ticket, String reason) throws Exception {
MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.FAILED);
assertEquals(reason, view.detail());
}
private MessageService.TaskView awaitTicketPhase(String ticket, MessageService.Phase phase) throws Exception {
long deadline = System.currentTimeMillis() + 3000;
MessageService.TaskView view;
do {
view = messages.poll(ticket);
if (view.phase() == phase) {
return view;
}
Thread.sleep(5);
} while (System.currentTimeMillis() < deadline);
assertEquals(phase, view.phase());
return view;
}
}
@@ -114,17 +114,6 @@ class RendezvousTest {
"the first resolution wins; the stored value is unchanged");
}
@Test
void resolveExhaustedTwiceIsANoOpTheSecondTime() {
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.open(W);
assertTrue(rendezvous.resolveExhausted(waiter, "first reason"), "the first classification resolves");
assertFalse(rendezvous.resolveExhausted(waiter, "second reason"),
"a second exhausted resolution on an already-resolved waiter returns false");
assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind());
assertEquals("first reason", waiter.getNow(null).text(),
"the first resolution wins; the stored value is unchanged");
}
@Test
void closeAskRemovesTheTurn() {
Rendezvous.AskTicket t = rendezvous.openAsk(W);
@@ -1,20 +0,0 @@
package dev.ltms.bridged.msg;
/**
* Explicit unbound tokens for tests that exercise delivery without an accepted send.
*
* <p>The waiter is {@code null} on purpose. "No accepted send" is an <em>absence</em>, and a helper
* that handed back a fresh {@code CompletableFuture} would invent one — which is how the first
* version of this class turned {@code captureBaselineSkipsTheReadWhenNoSendIsWaiting} red: the
* resolver saw a non-null waiter, decided a turn was in flight, and scraped a pane that no send was
* blocked on. An inert value must omit the fact, never fabricate it.
*/
public final class TestTurnTokens {
private TestTurnTokens() {
}
/** A token for a delivery that no send is waiting on: it authorises nothing. */
public static TurnToken inert(String target) {
return new TurnToken(target, null);
}
}
@@ -29,7 +29,6 @@ public final class FakeWorktrees implements Worktrees {
private final Set<String> existingPaths = ConcurrentHashMap.newKeySet();
private final Set<String> trackedPaths = ConcurrentHashMap.newKeySet();
private volatile RuntimeException addFailure;
private volatile boolean dirty = false;
private volatile String repoRoot = "/repo";
private volatile String prefix = "/worktrees";
@@ -62,12 +61,6 @@ public final class FakeWorktrees implements Worktrees {
return this;
}
/** Mark the worktree dirty so {@link #hasUncommitted} reports true (simulates uncommitted work). */
public FakeWorktrees withDirty(boolean dirty) {
this.dirty = dirty;
return this;
}
@Override
public String add(String repoRoot, String branch, String baseRef) {
addCalls.add(new AddCall(repoRoot, branch, baseRef));
@@ -84,11 +77,6 @@ public final class FakeWorktrees implements Worktrees {
removeCalls.add(new RemoveCall(repoRoot, worktreePath));
}
@Override
public boolean hasUncommitted(String worktreePath) {
return dirty;
}
@Override
public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
List<String> copied = new java.util.ArrayList<>();
@@ -175,46 +175,6 @@ class GitWorktreesTest {
assertTrue(Files.exists(Path.of(wt).resolve(".mcp.json")), ".mcp.json stub was dropped");
}
/**
* CB-576. {@code hasUncommitted} must treat a freshly-provisioned worktree as clean, but a
* worktree holding a brand-new, never-added file as dirty. The untracked-file-only shape is
* exactly the work lost in the incident — a worker's draft that compiled but was never
* committed because it stopped to ask its lead a question.
*/
@Test
void anUntrackedOnlyWorktreeCountsAsDirty(@TempDir Path tmp) throws Exception {
Path repo = initRepo(tmp.resolve("repo"));
GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
String wt = gitWorktrees.add(repo.toString(), "cb-576-u", "HEAD");
assertFalse(gitWorktrees.hasUncommitted(wt),
"a freshly provisioned worktree must read as clean");
Files.writeString(Path.of(wt).resolve("brand-new.txt"), "draft that was never added\n");
assertTrue(gitWorktrees.hasUncommitted(wt),
"an untracked-only file must count as dirty");
Files.writeString(Path.of(wt).resolve("README.md"), "edited tracked file\n");
assertTrue(gitWorktrees.hasUncommitted(wt),
"a tracked modification must also count as dirty");
}
/**
* CB-576 review. {@code hasUncommitted} must tolerate a missing worktree exactly like
* {@code remove}: an already-gone directory holds no work to lose, and throwing here would
* break teardown — SessionManager.release() calls it before stopping the pane, so an
* exception would orphan a live pane and skip the release notification (CB-516).
*/
@Test
void hasUncommittedOnAMissingWorktreeReturnsFalseWithoutThrowing(@TempDir Path tmp) {
GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString());
String gone = tmp.resolve("wts").resolve("does-not-exist").toString();
assertFalse(gitWorktrees.hasUncommitted(gone),
"a missing worktree is reported clean, not an error");
}
/** All three protected configs are covered: each one present in a worktree is neutralized and hidden. */
@Test
void allThreeConfigsAreNeutralizedWhenPresent(@TempDir Path tmp) throws Exception {
@@ -10,7 +10,6 @@ import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.herdr.WorkspaceControl;
import dev.ltms.bridged.member.ClaudeCodeLauncher;
import dev.ltms.bridged.msg.TestTurnTokens;
import dev.ltms.bridged.peer.PeerUnreachableException;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
@@ -102,7 +101,7 @@ class SessionManagerTest {
assertDoesNotThrow(() -> sessions.asPresence().markPresent(null),
"the primary's null terminal must not blow up an unrelated tool call");
assertDoesNotThrow(() -> sessions.onDelivered(null, TestTurnTokens.inert(null)));
assertDoesNotThrow(() -> sessions.onDelivered(null));
assertDoesNotThrow(() -> sessions.onTurnComplete(null));
assertDoesNotThrow(() -> sessions.onTurnFailed(null));
@@ -122,7 +121,7 @@ class SessionManagerTest {
"MCP presence moves SPAWNING → READY");
assertTrue(sessions.asPresence().isPresent(terminal), "presence is also recorded");
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
"delivery moves READY → BUSY");
@@ -154,7 +153,7 @@ class SessionManagerTest {
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnFailed(terminal);
@@ -182,7 +181,7 @@ class SessionManagerTest {
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnFailed(terminal);
@@ -265,7 +264,7 @@ class SessionManagerTest {
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
clock[0] = 100;
assertEquals(0, sessions.reapIdle(10), "BUSY session past TTL is never reaped");
@@ -282,7 +281,7 @@ class SessionManagerTest {
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnComplete(terminal);
clock[0] = 21;
@@ -300,7 +299,7 @@ class SessionManagerTest {
MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "owner2");
sessions.asPresence().markPresent(ready.terminalId());
sessions.asPresence().markPresent(busy.terminalId());
sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
sessions.onDelivered(busy.terminalId());
clock[0] = 50;
assertEquals(1, sessions.reapIdle(30), "only READY past TTL is reaped");
@@ -318,9 +317,9 @@ class SessionManagerTest {
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnComplete(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnComplete(terminal);
MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -338,13 +337,13 @@ class SessionManagerTest {
String terminal = session.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnComplete(terminal);
assertEquals(MemberSession.State.DONE,
sessions.get(session.paneId()).orElseThrow().state(),
"first turn completes without release");
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
sessions.onDelivered(terminal);
sessions.onTurnComplete(terminal);
assertTrue(sessions.get(session.paneId()).isEmpty(), "session released after cap reached");
@@ -360,7 +359,7 @@ class SessionManagerTest {
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
sessions.asPresence().markPresent(session.terminalId());
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
sessions.onDelivered(session.terminalId());
assertTrue(sessions.onTurnCompleteWithPostAction(session.terminalId()));
MemberSession updated = sessions.get(session.paneId()).orElseThrow();
@@ -374,7 +373,7 @@ class SessionManagerTest {
SessionManager sessions = sessionManager(herdr, () -> 0L, 1, true);
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
sessions.asPresence().markPresent(session.terminalId());
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
sessions.onDelivered(session.terminalId());
assertFalse(sessions.hasPostTurnAction(session.terminalId()),
"a session at its cap will be released, not reset for reuse");
@@ -389,7 +388,7 @@ class SessionManagerTest {
SessionManager sessions = sessionManager(herdr, () -> 0L, 0, false);
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
sessions.asPresence().markPresent(session.terminalId());
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
sessions.onDelivered(session.terminalId());
sessions.onTurnComplete(session.terminalId());
@@ -407,7 +406,7 @@ class SessionManagerTest {
MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "ownerB");
sessions.asPresence().markPresent(ready.terminalId());
sessions.asPresence().markPresent(busy.terminalId());
sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
sessions.onDelivered(busy.terminalId());
sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
@@ -6,15 +6,9 @@ import dev.ltms.bridged.guard.SubscriptionGuard;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.FakeHerdr;
import dev.ltms.bridged.herdr.WorkspaceControl;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import dev.ltms.bridged.member.ClaudeCodeLauncher;
import dev.ltms.bridged.msg.TestTurnTokens;
import dev.ltms.bridged.peer.MemberRole;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
@@ -184,49 +178,6 @@ class WorktreeSessionManagerTest {
assertTrue(sessions.get(paneId).isEmpty(), "released session is no longer retrievable");
}
/**
* CB-576. A normal {@code COMPLETED} release whose worktree holds uncommitted work must NOT
* remove it — {@code --force} would destroy the worker's only copy. The bridge cannot see
* uncommitted files, so the worktree is preserved and the release logged at WARN naming the
* path, the session, and the cause an operator needs to find the work.
*/
@Test
void releasePreservesDirtyWorktreeAndLogsWarn() {
FakeHerdr herdr = new FakeHerdr();
FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
.withDirty(true);
SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-576", null));
LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
ch.qos.logback.classic.Logger sessionLog =
(ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.setContext(ctx);
appender.start();
sessionLog.addAppender(appender);
sessionLog.setLevel(Level.WARN);
try {
sessions.release(s.paneId());
assertTrue(herdr.called("pane.close"), "release still tears the worker pane down");
assertTrue(worktrees.removeCalls().isEmpty(),
"a dirty worktree is never removed — it holds the only copy of the work");
String warn = appender.list.stream()
.filter(e -> e.getLevel().equals(Level.WARN))
.map(ILoggingEvent::getFormattedMessage)
.filter(m -> m.contains("dirty worktree"))
.findFirst()
.orElse("no dirty-release WARN logged");
assertTrue(warn.contains(s.worktree()), "the WARN names the worktree path: " + warn);
assertTrue(warn.contains(s.terminalId()), "the WARN names the session: " + warn);
assertTrue(warn.contains("COMPLETED"), "the WARN names the release cause: " + warn);
} finally {
sessionLog.detachAppender(appender);
}
}
@Test
void drainAllPreservesWorktreeOfIdleSession() {
FakeHerdr herdr = new FakeHerdr();
@@ -253,7 +204,7 @@ class WorktreeSessionManagerTest {
new WorktreeRequest("cb-544", null));
String terminal = s.terminalId();
sessions.asPresence().markPresent(terminal);
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal)); // BUSY, never completes → still BUSY when the timeout hits
sessions.onDelivered(terminal); // BUSY, never completes → still BUSY when the timeout hits
sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
+1 -4
View File
@@ -16,9 +16,6 @@
AuditLogTest, which attaches its own ListAppender and asserts on emitted records.
-->
<!-- Keep the test logger behaviour aligned with the production cancellation filter. -->
<turboFilter class="dev.ltms.bridged.logging.McpCancelledNotificationFilter"/>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
@@ -37,4 +34,4 @@
<appender-ref ref="STDOUT"/>
</root>
</configuration>
</configuration>
-937
View File
@@ -1,937 +0,0 @@
# M4 - Fleet health, recovery, routing, and capacity
**Status:** Design accepted on 2026-08-15. CB-573 part 1 has shipped the classification model and
the `bridge_list` capacity view; the remaining M4 units are not yet shipped.
**Scope:** Fleet evidence, safe mechanical repair, lead routing, capacity reporting, and optional
human notification.
**Grounded in:** `health/FleetHealth`, `health/PaneBudget`, `inject/StatusPoller`,
`inject/StatusRefiner`, `inject/CompletionResolver`, `inject/Injector`, `session/SessionManager`,
`msg/MessageService`, `msg/ReplyInbox`, `msg/ReplyPushLoop`, `msg/LeadHeartbeatLoop`,
`mcp/PrimaryRegistry`, and `herdr/AgentControl`.
## 1. Problem and decision boundary
The operator asked the bridge to detect idle agents, exceptions, stopped work, and broken
communication. The bridge may read an agent pane from time to time. It must notify a person when
the fleet cannot move forward.
The four operator terms are not four equal health states. `IDLE` is a normal mode. An exception is
sometimes visible only as pane text. Stopped work may look the same as slow work. Broken
communication can occur on several links.
M4 uses this boundary:
- The bridge detects facts and joins evidence.
- The bridge repairs only mechanical failures with no judgement.
- The lead decides whether to stop, retry, replace, or reassign a member.
- A human is notified only when no healthy lead can act.
- n8n may route an outbound incident. It never classifies state or chooses recovery.
An inbound n8n decider would need bridge authority. No narrow machine-decider role exists. Giving a
workflow engine lead authority is unsafe, while adding a new role is a separate authorization
design. An outbound sink needs no bridge role.
The bridge must never replay a delivered task. That task may already have changed files, pushed a
branch, opened a pull request, or changed external state. A replay can run those side effects twice.
This rule must remain true even if later code stores delivered prompt text.
## 2. Evidence model
A health state is mainly a comparison between two views:
- **herdr view:** current agents and raw live status from one `AgentControl.list()` call.
- **bridge view:** session FSM, MCP presence, accepted turns, tasks, inbox state, and lead ownership.
A strong fault often appears as a disagreement between those views. For example, `BUSY` in the
session FSM and `DONE` in herdr means the bridge missed a turn boundary. Pane reads support this
model, but they are not the main monitor.
`SessionManager.rosterView` already joins session state and live status. `AgentControl.list()`
already gets the whole live fleet in one call. M4 makes that join persistent and adds timers,
accepted-turn state, and incident state.
### 2.1 Real traces behind the design
The first trace was an architect that stopped making progress:
```text
profile=opus role=architect state=busy liveStatus=done
```
The session moved from `DONE` to `BUSY` for turn 2. Eighteen minutes later, the session still said
`BUSY`, herdr still said `DONE`, the async task still said `PENDING`, and no completion fallback had
run. This is `TURN_BOUNDARY_LOST`, not a general slow-turn guess.
The second trace had two async sends to the same pane, one second apart. The pane was then stopped.
One ticket became failed. The other stayed `pending - worker unknown`. Current
`MessageService.abandon` resolves only `Rendezvous.currentWaiter(target)`, while async tasks live in
a separate ticket map. CB-568 is intended to fix that bug. M4 still keeps an independent
post-teardown invariant so a later regression becomes `DELEGATION_ORPHANED`.
### 2.2 Corrections made during design
The first state table missed `BUSY` in bridged plus `IDLE` or `DONE` in herdr. It would have found
the real trace only through a late, weak stall timer. The final model adds
`TURN_BOUNDARY_LOST` as a strong disagreement state.
The first notification design also required a webhook before `health.enabled` could turn on. That
removed useful local detection to avoid a narrower human-notification gap. The final design splits
detection from notification. Missing human escalation is shown as partial coverage instead of
disabling health.
## 3. Classification precedence
Evidence is applied in this order. A lower rule cannot hide a higher one.
1. **Control link:** failed fleet list plus failed ping becomes `CONTROL_LINK_DOWN`.
2. **Definitive target loss:** `_not_found` becomes `GONE` or `LEAD_UNREACHABLE` when the control
link is healthy.
3. **Startup and teardown invariants:** readiness expiry becomes `NEVER_READY`; surviving tasks
after teardown become `DELEGATION_ORPHANED`.
4. **Bridge/live disagreement:** `BUSY` plus stable raw `IDLE` or `DONE` becomes
`TURN_BOUNDARY_LOST`.
5. **Known screen evidence:** a tested fatal signature becomes `ERROR_ON_SCREEN`.
6. **Timed suspicion:** unchanged sparse pane probes may become `STALL_SUSPECTED`.
7. **Communication quality:** completion fallback becomes `MUTE`; an old inbox entry becomes
`REPLY_STRANDED`.
8. **Normal mode:** `STARTING`, `IDLE`, `WORKING`, `WORK_PENDING`, or `BLOCKED_AMBIGUOUS`.
The member flow in Figure 1 shows lifecycle states and the main fault exits. Fault states are
reported beside the session FSM; most are not new FSM values.
```mermaid
flowchart TD
Registered["Member registered"] --> Starting["STARTING"]
Starting -->|"MCP presence"| Idle["IDLE"]
Starting -->|"Readiness grace expires"| NeverReady["NEVER_READY"]
Idle -->|"Accepted delivery"| Working["WORKING"]
Working -->|"Trusted turn boundary"| Idle
Working -->|"Bridge BUSY and herdr IDLE or DONE"| Lost["TURN_BOUNDARY_LOST"]
Working -->|"Known fatal screen"| Error["ERROR_ON_SCREEN"]
Working -->|"Long age and unchanged sparse probes"| Stall["STALL_SUSPECTED"]
Working -->|"Target not found"| Gone["GONE"]
Idle -->|"Inbox or queued delivery exists"| Pending["WORK_PENDING"]
Pending -->|"Delivery or collection finishes"| Idle
Idle -->|"Raw BLOCKED with an open turn"| Blocked["BLOCKED_AMBIGUOUS"]
Lost -->|"Strict guarded repair"| Repaired["DONE with reconciled completion"]
Lost -->|"Repair refused"| LeadDecision["Lead decision required"]
```
*Figure 1. The member lifecycle and the main health exits. Pane-based states never authorise an
automatic retry of the task.*
## 4. State model
### 4.1 Normal and transitional member states
| State | Exact evidence | Meaning and certainty |
|---|---|---|
| `STARTING` | Session is `SPAWNING`; MCP presence is absent | Normal inside the startup grace. MCP contact is the readiness signal. |
| `IDLE` | Session is `READY` or `DONE`; live status is `IDLE` or `DONE`; no open turn or inbox item exists | Normal. Idle is not a fault. |
| `WORKING` | Session is `BUSY`; raw live status is `WORKING`; the accepted turn is open | Certain that herdr sees work. It does not prove useful progress. |
| `WORK_PENDING` | Queued delivery or inbox content exists while the target is injectable | Transitional. Existing injector or push logic should move it. |
| `BLOCKED_AMBIGUOUS` | An open turn exists and raw live status is `BLOCKED` | The bridge cannot tell whether this is permission, input, or a settled screen. |
Idle may drive configured resource cleanup. It never opens an incident and never pages a person.
### 4.2 Member fault and quality states
| State | Exact evidence | Certainty and action |
|---|---|---|
| `NEVER_READY` | `SPAWNING`, no MCP presence, and an accepted delivery waits through the existing readiness grace | Delivery never became possible. The exact cause is unknown. Fail the send, stop the process, and preserve a provisioned worktree. |
| `GONE` | Per-target herdr call returns `_not_found` while fleet list or ping works | Certain target loss. Fail all target work. Do not replay it. |
| `TURN_BOUNDARY_LOST` | Same session turn stays `BUSY`; same accepted task stays open; two raw snapshots show `IDLE` or `DONE` | Strong disagreement. Strict reconciliation may repair it. |
| `ERROR_ON_SCREEN` | Suspicious non-working state survives grace; `detection` matches a tested adapter-specific fatal signature | Certain only for the matched signature. A bare word such as `Exception` is not enough. |
| `STALL_SUSPECTED` | Open turn is older than the configured threshold; two normalised `recent_unwrapped` digests are unchanged; no boundary or reply occurs | Not certain. A long valid API call can look the same. Lead decides. |
| `MUTE` | Turn resolves through completion fallback instead of `bridge_reply` | Certain that no structured reply won. It does not prove an MCP failure. A single event is a metric, not an incident. |
| `REPLY_STRANDED` | Typed reply or health message remains after owning-lead push reaches its cap | Collection failed. This does not explain whether the lead is busy, dead, or ignoring the nudge. |
| `DELEGATION_ORPHANED` | Target is gone, failed, or released, but one or more tasks remain `PENDING` after reconciliation grace | Certain bridge invariant failure. This is not an inbox-drain fault. |
| `WORK_PRODUCT_AT_RISK` | Provisioned branch has commits after its recorded base; member is `DONE`, `FAILED`, or preserved after release; no turn or inbox item remains; long-idle threshold passed | A warning, not proof of loss. Work may already have an open pull request or a squash merge. |
`MUTE` opens an incident only after a small fixed rate threshold for one target or profile, or when
it appears with another fault.
`WORK_PRODUCT_AT_RISK` must not become `WORK_PRODUCT_UNCOLLECTED`. The bridge does not know pull
request or merge state. If committed work appears with `REPLY_STRANDED` or
`DELEGATION_ORPHANED`, the existing incident gains `committedWorkAtRisk: true`.
### 4.3 Control-link state
| State | Exact evidence | Certainty and action |
|---|---|---|
| `CONTROL_LINK_DOWN` | Two full-fleet `agent.list` calls fail across the grace, and herdr `ping` also fails | Certain for the bridged-to-herdr link. Retry calls, record the incident, and use human escalation if no lead can be reached. |
A failed fleet list alone is not a dead-member claim. A single `_not_found` with a healthy global
link is a target fault, not a control-link fault.
### 4.4 Lead states
| State | Exact evidence | Meaning and action |
|---|---|---|
| `LEAD_IDLE` | Expected lead is present with raw injectable status; no actionable state waits | Normal. Existing heartbeat may run under its own policy. |
| `LEAD_WORKING` | Expected lead is present with raw `WORKING`; stall threshold is not met | Reachable and busy. Never inject into the live turn. |
| `LEAD_STATUS_UNKNOWN` | Expected lead is present with raw `UNKNOWN` | Neither dead nor a healthy routing target. Retain evidence and retry. |
| `LEAD_UNREACHABLE` | Expected lead is absent from two successful live-agent snapshots while ping works, or targeted lookup returns `_not_found` with a healthy control link | Route to a healthy peer. If none exists, use human escalation. |
| `LEAD_UNRESPONSIVE` | Actionable state waits; lead stays injectable; bounded nudges exhaust; inbox remains uncollected | Route to a healthy peer or a person. |
| `LEAD_STALL_SUSPECTED` | Lead stays `WORKING` past threshold; two sparse pane probes show no progress | Not certain. Never kill or restart automatically. Route to peer or person. |
The monitor retains the lead name and terminal, last successful sighting, raw status and age,
consecutive list absences, targeted errors, pane-probe facts, pending incident age, and nudge
outcomes. Current heartbeat and push loops discard much of this history.
Expected lead identity comes from the same supplier used by `CallerResolver`. It is not liveness
evidence. `LeadTabScanner` keeps cached identity after a failed scan, so the health monitor compares
that identity with a fresh successful agent list. A dynamic identity also survives a two-successful-
snapshot retirement grace. This stops a dead lead from escaping health by disappearing from one map.
### 4.5 Evidence limits
M4 cannot tell these cases apart with current evidence:
- A valid long call and a hung call may have the same status and pane digest.
- `BLOCKED` does not explain which input is needed.
- An idle prompt after failure may look like an idle prompt after success.
- A missing structured reply does not prove a broken MCP connection.
- An undrained inbox does not explain why the lead did not collect it.
- Arbitrary pane text cannot safely classify arbitrary exceptions.
- A branch ahead of its base does not prove that work was not collected.
Logs are outputs, not classifier inputs. The monitor never parses its own logs.
## 5. Automatic action and lead action
### 5.1 Actions the bridge may take
The bridge may:
- retry transient herdr status, list, ping, and pane-read failures with bounded backoff;
- re-submit Enter after the existing paste/submit race;
- fail queued delivery after `NEVER_READY`;
- stop a never-ready process while preserving its provisioned worktree;
- fail all queued, accepted, and async tasks for a gone or released target;
- reconcile one lost boundary when every strict gate in Section 8 passes;
- hold typed messages, nudge the owning lead, and stop at the configured cap;
- use the existing bounded idle-lead heartbeat;
- deduplicate, route, update, and resolve incidents.
These actions do not choose new work and do not replay old work.
### 5.2 Decisions reserved for the lead
Only the lead may:
- stop or continue `BLOCKED_AMBIGUOUS`;
- stop, inspect, or wait on `ERROR_ON_SCREEN`;
- kill or continue `STALL_SUSPECTED`;
- spawn a replacement or reassign work;
- retry a delivered task;
- choose how to use partial work in a worktree;
- restart herdr or change network, model, credentials, backend, or configuration.
Reports include literal safe tool calls such as `bridge_status(sessionId="...")`,
`bridge_poll(ticket="...")`, `bridge_list()`, and optional `bridge_stop(paneId="...")`. A judgement
state never presents stop as the only action.
### 5.3 Release causes and worktree safety
| Release cause | Process action | Provisioned worktree |
|---|---|---|
| `SPAWN_ROLLBACK` before registration or delivery | Stop and clean up | Remove |
| `COMPLETED` for `READY` or `DONE` without pending work, idle TTL, or successful context-cap completion | Stop | Remove only if clean; preserve a dirty worktree (CB-576) |
| `NEVER_READY` | Stop | Preserve |
| `GONE` | Best-effort stop | Preserve |
| `TURN_FAILED` or lead abort while `BUSY` or `FAILED` | Stop | Preserve |
| `RELEASE_WITH_PENDING_TASKS` | Stop | Preserve |
| `SHUTDOWN` | Stop | Preserve |
Explicit stop is state-aware. `SPAWNING`, `BUSY`, `FAILED`, or any target with pending tasks uses a
preserving cause.
Before abnormal release removes the live session, M4 writes an atomic manifest under the worktree
root. It records session identity, owner, role, profile, repository, path, branch, base commit,
release cause, release time, state, and pending task ids. `bridge_list.preservedWorktrees` loads these
manifests after restart. Stop output and WARN logs also name the path and cause. M4 never
auto-deletes a preserved worktree.
## 6. Fleet health monitor
Add `FleetHealthMonitor`. Do not widen `StatusPoller` into a policy loop.
`StatusPoller` has a 250 ms delivery cadence and samples only injector targets with outstanding
work. Health needs all sessions, all leads, task state, inbox age, and global control evidence. One
loop cannot serve both cadences safely.
Build the monitor like `LeadHeartbeatLoop`:
- pure `decide(snapshot, priorState, now)` logic;
- a thin scheduler;
- an injected clock;
- edge-triggered state changes;
- no network work in the pure function;
- no sleeping in tests.
Each enabled fleet tick reads:
- one `AgentControl.list()` result for the whole fleet;
- one in-memory `SessionManager.roster()` snapshot;
- accepted turns and async task state;
- typed inbox depth, kind, and age;
- push and heartbeat outcomes;
- configured and discovered leads.
Existing failure paths publish structured evidence to the monitor. The monitor does not infer events
from log text.
### 6.1 Pane budget
Healthy idle members, recent working members, and quiet leads cause no pane reads.
A pane is eligible only for a stable lost boundary, sustained `BLOCKED` or `UNKNOWN`, work older
than the suspect threshold, or one final evidence read for a confirmed fault when the pane exists.
Compiled brakes apply even if config asks for more:
- per-target pane cooldown is at least 60 seconds;
- working age before the first progress probe is at least 300 seconds;
- at most two pane reads occur in one fleet tick;
- targets rotate fairly;
- only a normalised digest and optional clipped local excerpt are stored;
- no pane excerpt leaves bridged in a human webhook.
Use `detection` for tested screen signatures. Use normalised `recent_unwrapped` only for progress
comparison.
## 7. Typed inbox and routing
### 7.1 Semantic record
The typed inbox record carries:
```text
schemaVersion
kind: reply | health
msgId, target, subjectTerminal, recipientLead
severity, state, evidence
createdAtEpochMillis, firstSeenEpochMillis, lastSeenEpochMillis
recoveryTried, suggestedToolCalls, content
```
A health message never calls `Rendezvous.resolve`. It cannot look like the member's task result.
Both inbox adapters share field preservation, first-id-wins dedup, FIFO among decoded messages,
explicit ownership, ack, and release rules. The in-memory adapter stores typed records directly. It
does not copy AMQP migration logic.
### 7.2 AMQP migration
The reader uses AMQP `content_type`, never body sniffing:
```text
Legacy v0: text/plain
Typed family: application/vnd.ltms.bridged.inbox-message+json
```
A legacy reply may begin with `{`. It remains plain text because its media type is `text/plain`.
Legacy text becomes `kind=reply` with exact UTF-8 content and absent typed metadata.
Typed JSON has required integer `schemaVersion: 1`. Version 1 ignores unknown optional fields.
Missing required fields, invalid enums, malformed UTF-8 or JSON, and property/body identity mismatch
are invalid data.
An unknown schema version is not partly decoded. It remains unacknowledged on the original queue and
creates one operator-visible `unsupported_version` failure. A newer daemon may read it later.
Invalid known-format data is copied byte-for-byte to durable queue
`agent.<target>.inbox.quarantine`. A dedicated confirm-mode publisher confirms the persistent copy
before the original is acknowledged. A failed quarantine handoff leaves the original unacknowledged.
The raw body never enters logs.
Decode failure creates a redacted WARN, metric, `bridge_list` summary, and routed health incident.
One bad entry never escapes the consumer callback and never stops later valid messages.
Safe downgrade is not supported. The previous build ignores `content_type` and would show typed JSON
as ordinary reply text. If drained, it would acknowledge the message and lose typed meaning. Typed
queues must be drained or preserved before an old jar runs.
The existing contract suite uses RabbitMQ. Production uses LavinMQ. The migration and lead-key
ownership cases must run once against production LavinMQ before release, or the release must state
that LavinMQ was not checked.
### 7.3 Member routing
A member incident first goes to the exact lead that owns its accepted delegation.
`PrimaryRegistry` needs a no-fallback `delegatingLeadFor(memberTarget)` query. Health routing must not
use the old singular-primary fallback when several leads exist.
Publish the incident under the affected member target. Trigger the existing bounded push route. The
push waits until the owning lead is injectable, so it does not interrupt a live lead turn.
### 7.4 Peer lead routing
Figure 2 shows the route from incident to lead, peer, or person.
```mermaid
flowchart TD
Incident["Open incident"] --> Member{"Member incident?"}
Member -->|"yes"| Known{"Exact delegation owner known?"}
Known -->|"no"| Sink{"Human webhook enabled and healthy?"}
Known -->|"yes"| Owner{"Owner lead healthy?"}
Owner -->|"yes"| OwnerInbox["Publish to owner lead path"]
Owner -->|"no"| PeerSet["Build healthy peer candidate set"]
Member -->|"no, lead incident"| PeerSet
PeerSet --> Peer{"Healthy peer exists?"}
Peer -->|"yes"| Select["Choose fewest assigned incidents<br/>then stable name and terminal id"]
Select --> PeerInbox["Publish to peer lead inbox<br/>and status-gated push"]
Peer -->|"no"| Sink
Sink -->|"yes"| Webhook["Send classified outbound incident"]
Sink -->|"no"| Passive["Keep incident open<br/>show partial coverage on local surfaces"]
```
*Figure 2. Routing keeps delegation ownership separate from temporary peer fallback.*
Peer candidates exclude the incident subject, failed owner, absent leads, raw-unknown leads, and
leads with an open unhealthy state. A reachable `WORKING` peer may be selected; its push waits for an
injectable window.
Choose the candidate with the fewest assigned foreign incidents. Break ties by stable lead name,
then terminal id. Pin the recipient. Reassign only if that peer becomes unhealthy or retires. A
routing generation marks a reassignment, and old pending assignments become superseded.
`bridge_list` lead rows show health, health age, assigned foreign incident count, and a bounded list
of incident id, subject, state, severity, age, and routing generation. The top-level view also shows
owner, recipient, and routing reason.
A peer incident is published under the recipient lead's inbox key, not the failed subject's key. Its
status-gated nudge names the failed lead and gives the exact
`bridge_poll(target="<recipient-terminal>")` call.
### 7.5 Lead inbox ownership
Add `LeadInboxRegistry`, driven by the same expected-lead supplier as `CallerResolver`.
It calls `replyInbox.own(leadTerminal)` at startup for configured leads, after successful discovery,
after config adds a lead, and before publication. Ownership is not an authorization side effect.
A missing lead keeps its key owned. Release happens only after confirmed retirement, all incidents
are reassigned or resolved, typed health messages move or ack, and the queue is empty. Own a
replacement terminal before moving messages from the old key. Never release a non-empty in-memory
lead key, because in-memory release clears local data.
### 7.6 Single-lead deployment
One lead and no peer is a normal mode, not an edge case.
An idle, reachable lead may receive the existing bounded nudge. An unreachable or stalled sole lead
has no safe in-loop recovery. The bridge must not restart or replace it. A new lead would not have the
failed lead's plan or context, and an uncertain relaunch could create two orchestrators.
With no webhook, only `bridge_list`, `/healthz`, metrics, WARN logs, and the incident journal remain.
These are passive surfaces. They are not a human notification.
## 8. Lost-boundary reconciliation
This is the only M4 path that reconstructs a result. It must prefer a visible stall over a fabricated
reply.
### 8.1 Why normal completion rules are not enough
Current `CompletionResolver.resolve` has two fail-open rules. It resolves when the delivery baseline
is missing. It also resolves an empty completion when the pane read fails. Those choices are valid
after a trusted `WORKING -> IDLE` boundary because the bridge knows the turn ran. They are unsafe
when health only guesses that a boundary was lost.
M4 gives each accepted send an internal `TurnToken`. It ties target, exact waiter, session turn,
delivery baseline, and task outcome together.
### 8.2 Delivery baseline
Capture the baseline immediately after prompt send and before the delivery future completes. Store:
```text
TurnToken
exact waiter identity
capture time and pane source
normalised assistant block clipped to MAX_SCRAPE_CHARS
whether a supported assistant marker was recognised
capture result: PRESENT | READ_FAILED | UNRECOGNISED
```
A failed or missing baseline never authorises repair. A late baseline is not valid evidence. After a
daemon restart, the old waiter, task, token, and baseline are gone, so the old turn cannot be
repaired.
Automatic repair is enabled only for agent kinds with tested assistant-block fixtures. Current
extraction is Claude Code-specific and falls back to arbitrary raw text without `⏺`. That raw fallback
cannot authorise repair. OpenCode repair stays disabled until live pane fixtures exist.
### 8.3 Strict gates and resolver result
Figure 3 shows the repair gates. Any failed gate keeps the waiter unchanged.
The two raw snapshots must describe the same `TurnToken` and session turn. No `WORKING`,
`BLOCKED`, `UNKNOWN`, missing-agent, reply, failure, or new-delivery observation may occur between
them.
```mermaid
flowchart TD
Candidate["TURN_BOUNDARY_LOST candidate"] --> Stable{"Same TurnToken and BUSY turn<br/>across two raw IDLE or DONE snapshots?"}
Stable -->|"no"| Resnapshot["Take a fresh snapshot"]
Stable -->|"yes"| Waiter{"Exact captured waiter<br/>still open by identity?"}
Waiter -->|"no"| Stale["STALE_TURN or ALREADY_RESOLVED"]
Waiter -->|"yes"| Baseline{"Successful recognised<br/>delivery baseline exists?"}
Baseline -->|"no"| Refuse["Refuse repair<br/>leave ticket pending"]
Baseline -->|"yes"| Read{"Fresh pane read succeeds?"}
Read -->|"no"| Refuse
Read -->|"yes"| Output{"Recognised non-blank assistant block<br/>differs from clipped baseline?"}
Output -->|"no"| Refuse
Output -->|"yes"| Resolve["Shared CompletionResolver guard core<br/>resolves exact waiter"]
Resolve -->|"won race"| Repaired["RECONCILED_COMPLETION<br/>same turn becomes DONE"]
Resolve -->|"lost race"| Resnapshot
```
*Figure 3. Repair needs stronger evidence than a normal observed turn boundary.*
Refactor the current resolver into one guard core with two policies:
```text
resolveCaptured(target, inFlight, OBSERVED_BOUNDARY)
resolveCaptured(target, inFlight, LOST_BOUNDARY_REPAIR)
```
The health monitor calls only:
```text
CompletionResolver.reconcileLostBoundary(target, expectedTurnToken)
```
It returns `REPAIRED`, `ALREADY_RESOLVED`, `REFUSED_NO_CAPTURE`, `REFUSED_NO_BASELINE`,
`REFUSED_UNREADABLE`, `REFUSED_UNCHANGED`, `REFUSED_AMBIGUOUS_OUTPUT`, `STALE_TURN`, or
`RACE_LOST`.
Only `REPAIRED` and same-turn `ALREADY_RESOLVED` may move that turn from `BUSY` to `DONE`. A
per-target reconciliation gate stops a queued second send from being accepted between waiter
resolution and the FSM transition.
### 8.4 Lead-visible marker and refusal
A repaired result uses distinct `RECONCILED_COMPLETION` values in `Rendezvous`, `MessageService`,
task poll source, and metrics. The lead sees:
```text
[repaired completion - bridged detected a lost turn boundary. The member did not call
bridge_reply; pane-derived text follows and may be partial]
```
Clipped text also keeps the existing clipped-tail marker.
A refused repair leaves `TURN_BOUNDARY_LOST` open and the ticket pending. The report states that no
reply was reconstructed and no task was replayed. `UNCHANGED`, `UNREADABLE`, and
`AMBIGUOUS_OUTPUT` get at most one delayed retry for the same token. Missing capture or baseline gets
no retry. After two refused scrapes, automatic repair stops for that token.
### 8.5 Target-wide teardown invariant
CB-568 owns the multi-ticket cancellation mechanism. M4 routes every terminal cause through that one
idempotent operation and checks this independent invariant after teardown:
- no injector entry exists for the target;
- no accepted turn or completion record exists;
- no rendezvous waiter or ask exists;
- every async task is terminal or was already terminal;
- no thread waiting for the target send lock can later accept it;
- new sends fail immediately;
- each old task has one terminal outcome and one metric count.
A violation becomes `DELEGATION_ORPHANED`. The monitor may call the same idempotent target-wide
failure operation once. It never recreates the task.
## 9. Capacity and utilisation
Capacity is a view, not a health state.
`bridge_list` adds one block per profile:
```text
profile, maxLoad, live, free, reclaimable
```
For an unlimited profile, `maxLoad` and `free` are null. `free` is
`max(0, maxLoad - live)` for a capped profile.
The view must use the exact live-count function used by placement. A second calculation could show a
free slot that placement then refuses. Member rows add `idleForSeconds` only when state is `READY` or
`DONE`, no accepted turn exists, and the inbox is empty. `reclaimable` means only that the member
holds capacity without open bridge work.
The existing idle-lead nudge gains a bounded capacity summary. It lists per-profile live, cap, free,
and reclaimable counts, plus at most three long-idle members. Capacity does not make
`FleetState.hasPending()` true. A changed capacity fingerprint may re-arm one capped heartbeat
sequence. The fingerprint excludes changing idle durations, so a static idle fleet cannot reset the
cap forever. Reply-push stand-down remains first.
The bridge must never:
- spawn a member because a slot is free;
- generate a task or acceptance criteria;
- move queued work to another member or profile;
- treat a free slot or idle member as an incident;
- stop an idle member only to improve utilisation.
The bridge knows capacity facts but has no work list. Only the lead has the plan, task context,
side-effect history, and acceptance criteria.
Capacity calculation is in memory and adds no pane reads. Work-product checks run on a terminal
session edge, not every fleet tick.
This capacity design adds no automatic stop. The accepted `NEVER_READY` cleanup can still stop a
very slow startup after the existing grace, which is a known risk. Free capacity and long idle time
never trigger that path.
## 10. Human escalation and notification
### 10.1 Escalation rule
Notify a person only when no healthy lead can act:
- `CONTROL_LINK_DOWN` survives grace;
- a lead is unhealthy and no healthy peer can receive the incident;
- a member incident has no known owning lead;
- the only owning lead becomes unreachable, unresponsive, or stalled;
- incident publication or routing itself fails.
Do not page a person for a member fault while a healthy owning lead exists. An uncollected member
incident feeds lead-health evidence. If the lead then becomes unhealthy, peer or human routing starts.
### 10.2 Detection and notification switches
`health.enabled` controls detection and bridge-local reporting. It does not require a webhook.
`health.notifications.mode` is `disabled` or `webhook`. Disabled is valid and is the default.
Webhook mode requires a resolved environment variable. Turning notification off stops outbound
attempts but keeps incidents. Turning it back on resumes still-open human incidents.
Without a sink, `bridge_list.healthCoverage` states that human escalation is unavailable. `/healthz`
keeps its existing HTTP liveness result and adds a nested `fleetHealth.status=partial` component.
Metrics and one startup or reload WARN expose the same limit.
### 10.3 Incident and delivery deduplication
One open incident uses this key:
```text
(scope, subjectStableId, state, causeFingerprint)
```
The cause fingerprint includes stable error codes, dependency names, signature ids, or invariant
names. It excludes times, ages, retry counts, pane text, and changing digests. A later recurrence
after resolution gets a new generation and incident id.
Each outbound event uses:
```text
Idempotency-Key = hash(incidentId, eventType, eventRevision)
```
Event types are `open`, `severity_changed`, `reminder`, and `resolved`. Transport retries keep the
same key.
An atomic owner-only journal beside the active config stores open incidents, routing, delivered
revisions, retry state, and resolution state. It stores no pane or task content. Journal failure does
not stop detection, but notification coverage becomes degraded.
### 10.4 Retry, reminder, and resolve
Send the first event immediately. Retry network errors, timeouts, HTTP 408, HTTP 429, and HTTP 5xx
with full-jitter exponential backoff:
```text
base: 5 seconds
factor: 3
maximum delay: 15 minutes
one outstanding attempt per event
```
Respect `Retry-After` up to 15 minutes. Other HTTP 4xx responses are permanent for that event until
config changes or a person requests replay.
Transport retry is not an incident reminder. `humanRepeatSeconds` creates a new reminder revision
for an unresolved critical incident after the last successful human event. Disabled mode does not
build an unbounded reminder queue.
Send `resolved` only if at least one human event for that incident was delivered. If an incident
resolves before its first successful delivery, cancel the pending open event and record local
resolution.
### 10.5 Outbound payload boundary
An outbound payload may contain incident id and event type, severity, state, scope, stable bridge
ids, role or profile, times, duration, structured evidence type and counts, recovery attempted,
routing reason, coverage, and safe tool calls.
It must never contain:
- raw pane text, pane excerpts, or pane digests;
- task briefs, prompts, or member reply content;
- source files, diffs, or worktree file content;
- worktree paths;
- environment values, tokens, credentials, headers, or webhook URL;
- raw exception messages or stack traces;
- arbitrary model output.
The sink response body is ignored. A webhook cannot direct recovery. n8n remains outbound-only.
### 10.6 Metrics
M4 adds bounded-label series:
```text
bridged_health_incidents{scope,state,severity}
bridged_health_incidents_total{event}
bridged_health_notifications_total{event,outcome}
bridged_health_notification_queue_depth
bridged_health_notification_last_success_seconds
bridged_health_notification_capability{mode,status}
bridged_lead_health{lead,state}
bridged_lead_assigned_incidents{lead}
```
Metric labels never include terminal ids, incident ids, URLs, or error text.
## 11. Configuration
The optional `health:` block is absent or disabled by default. The dormant monitor scheduler does no
herdr or pane work while disabled. Every listed key is hot because the monitor reads `ConfigRef` on
each tick or notification.
| Key | Class | Default and hard bound | Purpose |
|---|---|---|---|
| `health.enabled` | Hot | `false` | Enable detection and bridge-local reporting. |
| `health.snapshotIntervalSeconds` | Hot | default 30, minimum 15 | Whole-fleet comparison cadence. |
| `health.workingSuspectAfterSeconds` | Hot | default 600, minimum 300 | Age before working-pane probes. |
| `health.paneProbeIntervalSeconds` | Hot | default 60, minimum 60 | Per-target pane cooldown. |
| `health.leadUnresponsiveAfterSeconds` | Hot | default 300, minimum 120 | Delay after exhausted actionable nudges before lead fault. |
| `health.humanRepeatSeconds` | Hot | default 3600, minimum 900 | Minimum repeat period for one open human incident. |
| `health.capacityLongIdleAfterSeconds` | Hot | default 900, minimum 300 | Long-idle threshold for capacity summaries. |
| `health.includePaneExcerpt` | Hot | `false` | Allow a clipped excerpt in local lead reports only. Human payloads still exclude it. |
| `health.notifications.mode` | Hot | `disabled` | Select `disabled` or `webhook`. |
| `health.notifications.webhookUrlEnv` | Hot | required in webhook mode | Name of the environment variable that holds the sink URL. |
| `health.notifications.requestTimeoutMs` | Hot | default 10000, range 1000-30000 | Whole webhook request limit. |
Two consecutive snapshots are compiled floors for lost boundary, lead disappearance, and control
link failure. The two-pane-reads-per-tick limit is also compiled and cannot be weakened by config.
## 12. Delivery units and acceptance
### Unit 1 - Evidence model and fleet snapshot
Scope: health state model, fleet join, clocks, evidence retention, and pane budget.
Acceptance criteria:
1. One `agent.list` call covers one enabled fleet tick.
2. Pure decision tests cover every state and every evidence limit in Section 4.
3. `BUSY` plus stable raw `DONE` opens `TURN_BOUNDARY_LOST` after two snapshots.
4. Healthy fleet snapshots perform zero pane reads.
5. Pane cooldown, two-read fleet budget, and fair rotation cannot be disabled by config.
6. Logs are outputs only; no log parsing exists.
7. Fleet snapshots expose the same profile live-count calculation that placement uses.
8. Capacity rows report cap, live, free, and reclaimable values without opening incidents.
### Unit 2 - Lost boundary and task reconciliation
Scope: accepted-turn identity, guarded repair, target-wide teardown, release causes, and preserved
worktree discovery.
Acceptance criteria:
1. Every accepted send receives a stable `TurnToken` tied to target, exact waiter, and delivery
baseline.
**Corrected during implementation (2026-08-15).** This criterion first also required the session
turn number and the task outcome. That is not implementable at this layer, and the implementer
refused it three times rather than fabricate a value — correctly. The reason is an ordering fact
that is invisible from any single class: `MessageService` owns acceptance and holds the waiter and
the async `Task`, but it learns nothing about delivery, because the delivery event goes to
`CompletionResolver` through `TurnListener.onDelivered`. And `CompletionResolver.onDelivered` runs
*before* `SessionManager.onDelivered`, so the session turn number does not exist yet at the only
point where the token could capture it.
Two ways out were rejected. A shared registry keyed by target reintroduces exactly the "whichever
send happens to be waiting" ambiguity the token exists to remove — the same weak claim
`Rendezvous.currentWaiter` warns about. Injecting a turn counter into `MessageService` adds a
required cross-layer dependency to populate a field that nothing in this slice reads, which is
speculative coupling across a boundary already shown to be fragile.
So the token identifies the **accepted send**, and `SessionManager` keeps verifying its own
delivery separately. Repair (criterion 2) does need the session turn; binding it means resolving
that acceptance-versus-delivery ordering first, and that work belongs to the repair unit, not
here. The token record carries a comment saying the field is deliberately absent.
2. Repair requires the same `BUSY` token, two raw `IDLE` or `DONE` snapshots, no conflicting
observation, exact open waiter, successful baseline, and new recognised assistant output.
3. Missing, failed, late, or post-restart baseline never authorises repair.
4. Repair is enabled only for agent kinds with tested assistant-block extraction. Raw-text fallback
without a recognised marker refuses repair.
5. Normal completion and repair use one resolver guard core. Waiter, scrape, clipping, unchanged, and
exact-turn guards are not duplicated.
6. `reconcileLostBoundary` returns every typed result named in Section 8.3.
7. Only `REPAIRED` and same-turn `ALREADY_RESOLVED` may move the same turn to `DONE`.
8. A per-target reconciliation gate blocks a queued second send during repair and FSM update.
9. Repaired completion has distinct rendezvous kind, message outcome, poll source, lead marker, and
metric. Clipping keeps its extra marker.
10. Unchanged, unreadable, or ambiguous evidence gets at most one delayed retry. Missing capture or
baseline gets none.
11. Refusal leaves the ticket pending and tells the lead that no result was rebuilt or replayed.
12. Release, gone, never-ready, and abnormal stop use CB-568's one idempotent target-wide failure
operation.
13. The post-teardown invariant in Section 8.5 is tested independently of CB-568 internals.
14. A violated teardown invariant creates `DELEGATION_ORPHANED` and retries only the idempotent
failure operation.
15. `SPAWN_ROLLBACK` and normal `COMPLETED` remove worktrees. Abnormal and shutdown causes preserve
them.
16. Explicit stop is state-aware. Any pending task or non-terminal state preserves the worktree.
17. Atomic preserved-worktree manifests reload after restart and appear in lead-only
`bridge_list.preservedWorktrees`.
18. Manifest failure preserves the worktree and opens an operator-visible health failure.
19. Provision records the base commit. Terminal, long-idle worktrees report
`WORK_PRODUCT_AT_RISK` only under the evidence in Section 4.2 and never auto-delete work.
20. No path replays a delivered task, rebuilds its brief, or retargets it, even when prompt text is
available.
21. Tests cover both real traces, all repair refusals, clipping, explicit-reply and next-turn races,
restart without capture, concurrent send and release, and preserved discovery after restart.
### Unit 3 - Typed inbox and member routing
Scope: semantic record, AMQP migration, both adapters, member routing, polling, and member health in
`bridge_list`.
Acceptance criteria:
1. AMQP selects legacy or typed decoding only from `content_type`; it never sniffs the body.
2. Persistent `text/plain` from the old build becomes `kind=reply` with exact UTF-8 content,
including content beginning with `{`.
3. New entries use the vendor media type, `schemaVersion: 1`, UTF-8, persistent delivery, and AMQP
message ids.
4. Version 1 ignores unknown optional fields but rejects missing fields and identity mismatch.
5. Unknown versions are not decoded or acked. They remain on the original queue and create one
deduplicated failure.
6. Invalid known data never escapes the callback, appears as a reply, or blocks later valid messages.
7. Invalid data reaches durable per-target quarantine before original ack. Failed handoff leaves the
original unacked.
8. Decode failures create redacted WARN, metric, `bridge_list` summary, and routed incident without
raw content.
9. Both adapters pass one semantic contract for fields, FIFO, dedup, ownership, ack, and release.
10. Lead keys require explicit ownership. Publication never claims a queue.
11. Unit codec tests cover legacy `{`, Unicode, malformed UTF-8, typed round trip, additive fields,
malformed JSON, missing fields, identity mismatch, media type, version, and dedup.
12. A live broker contract writes old wire data and reads it with the new adapter after reconnect.
13. Live contract tests cover mixed entries, quarantine confirm-before-ack, unsupported redelivery,
later progress past poison, property persistence, lead ownership, and ack removal.
14. Safe downgrade is documented as unsupported.
15. RabbitMQ contract tests pass with `mvn test -Pcontract`. The same cases run once on production
LavinMQ, or the release states that LavinMQ was not checked.
16. Member incidents route to the exact delegating lead and never resolve a task rendezvous.
17. `bridge_list` shows compact member health and capacity without pane content. Member
`idleForSeconds` is present only when no accepted turn or inbox item exists.
### Unit 4 - Lead health and peer routing
Scope: lead evidence, exact ownership, peer selection, explicit-recipient push, and lead inbox
lifecycle.
Acceptance criteria:
1. Lead identity uses the `CallerResolver` supplier. Liveness uses successful current agent data.
2. Two successful-list absences with healthy ping become `LEAD_UNREACHABLE`; global link failure does
not.
3. Raw `WORKING`, raw `UNKNOWN`, first-seen time, failures, last success, and error class persist
across ticks.
4. Heartbeat and push publish status and nudge outcomes before safe no-injection decisions.
5. Dynamic lead identity survives a two-successful-snapshot retirement grace.
6. Member incidents first use exact delegation ownership with no singular-primary fallback.
7. Peer selection follows the exclusions, load rule, and stable tie break in Section 7.4.
8. A selected working peer is not interrupted. Its push waits for an injectable window.
9. Recipient assignment stays pinned. Reassignment increments generation and supersedes old pending
assignment.
10. `bridge_list` shows bounded foreign assignments, recipient, reason, and generation without pane
content.
11. `LeadInboxRegistry` owns configured and discovered lead keys before publication.
12. Missing leads keep ownership. Retirement needs an empty queue and handled incidents.
13. Replacement owns the new key before messages move. Non-empty in-memory keys are not released.
14. Tests cover dead versus busy, unknown, global failure, stale scan cache, disappearance, one peer,
several peers, reassignment, and no peer.
15. Adapter tests cover lead ownership, restart re-ownership, retirement, and terminal replacement.
LavinMQ is checked or named as unchecked.
16. A sole unreachable or stalled lead is never restarted or replaced. Without a sink, only passive
evidence remains and every coverage surface says so.
### Unit 5 - Human sink, hot config, metrics, and operator coverage
Scope: generic webhook, config split, incident journal, retry, resolve, metrics, example config, and
operator documentation.
Acceptance criteria:
1. `health.enabled` works without a human sink.
2. Notification mode is hot, defaults to disabled, and supports disabled or webhook.
3. Webhook mode requires a resolved environment value. Bad notification config does not disable an
already valid detector.
4. Mode changes keep open incidents. Re-enable resumes eligible incidents.
5. `bridge_list`, `/healthz`, metrics, and one WARN show partial coverage without a sink. HTTP
liveness behavior stays unchanged.
6. One-lead, no-sink coverage states that lead failure has no active notification or recovery.
7. Incident and outbound dedupe use the stable keys in Section 10.3.
8. The owner-only local journal survives restart and contains no pane or task content.
9. Journal failure keeps detection running but marks notification coverage degraded.
10. Retry tests cover network failure, timeout, 408, 429, `Retry-After`, 5xx, permanent 4xx, jitter,
delay cap, config re-arm, and one outstanding attempt.
11. Reminders and transport retries remain separate. Disabled mode does not build an unbounded queue.
12. Resolve sends only after an earlier human event succeeded. Resolve-before-delivery cancels stale
open delivery.
13. Metrics use bounded labels and exclude ids, URLs, and error text.
14. Payload tests reject every content type forbidden in Section 10.5.
15. Webhook response bodies are ignored and cannot direct recovery.
16. Tests cover disabled mode, one lead without sink, open/update/reminder/resolve, restart, dedup,
reassignment, disable/re-enable, and sink failure while local health continues.
17. `bridged.example.yaml` documents all hot keys and compiled floors.
18. The operator Features wiki is updated separately. The portable `CLAUDE.md` block is checked and
changed only if shipped tool or inbox semantics make it untrue.
19. `mvn clean install` passes.
## 13. Not checked and release gates
These limits are part of the design, not optional follow-up notes.
- **OpenCode pane status and assistant markers were not checked.** OpenCode lost-boundary repair is
disabled until live fixtures exist.
- **Permission-prompt status was not checked** for Claude Code or OpenCode. `BLOCKED` remains
ambiguous and has no automatic action.
- **`recent_unwrapped` stability was not checked** across all supported agent kinds. If normalisation
is not stable, `STALL_SUSPECTED` must say its evidence is weaker.
- **The real `BUSY + DONE` trace was not replayed against live herdr.** The design uses the observed
production trace and current poller behavior.
- **CB-568 was not present when Unit 2 was designed.** Unit 2 must inspect the landed API and keep its
independent teardown invariant.
- **Production LavinMQ was not checked.** Existing durable-inbox contracts use RabbitMQ. Migration,
quarantine, redelivery, lead ownership, and reassignment must run on LavinMQ before release or be
recorded as unchecked.
- **Live multi-lead routing was not checked.** Peer choice and reassignment are design rules backed by
fake-clock and adapter tests until a live exercise runs.
- **A live sole-lead failure with a webhook was not checked.** The no-peer path is a design result,
not a tested recovery.
- **No n8n, Slack, PagerDuty, or other receiver was checked.** The webhook remains generic and
outbound-only.
- **Deployment supervisor behavior for nested `/healthz` fields was not checked.** HTTP liveness
status stays unchanged to reduce this risk.
- **Incident-journal crash behavior was not checked** because the journal does not exist yet. Unit 5
must test atomic replacement and restart recovery.
- **Worktree merge state cannot be checked reliably** without forge or explicit collection evidence.
`WORK_PRODUCT_AT_RISK` stays a warning.
## 14. Locked exclusions
M4 does not expose `agent.read` as a bridge tool. It does not add a workflow engine, inbound n8n
authority, automatic task assignment, task replay, automatic lead replacement, or automatic member
spawn for free capacity.
The bridge remains a message bus with evidence and bounded mechanical repair. The lead remains the
place where judgement and work planning happen.