Compare commits

..

5 Commits

Author SHA1 Message Date
Dai Ha edabccd885 CB-637: relabel lead-comms (was colliding CB-635) + document coordId route in the CLAUDE.md tool table
The lead-to-lead wiring shipped with CB-635 in its comments, but CB-635 is
already the broker.uriEnv / unreachable-broker work. Relabel the mailbox +
fleet_send{coordId} + receive loop to CB-637 so a ticket number names one
feature. Add the cross-host peer-lead row to the primary intent->tool table
(kept byte-identical with the wiki template).
2026-08-24 17:48:51 +02:00
Dai Ha 1dbe3a03fc Merge lead-comms-wiring: wire LeadMailbox into the daemon (fleet_send{coordId} + receive loop) 2026-08-24 17:42:11 +02:00
Dai Ha 6058b8472b lead comms: wire LeadMailbox into the daemon so lead-to-lead messages flow
CI / contract (pull_request) Successful in 41s
CI / build (pull_request) Failing after 1m22s
The sibling ticket landed the mechanism (LeadMailbox, LeadMessage, the
coordinator: config block) but nothing opened it, nothing sent through it, and
nothing read it. This is the wiring.

- LeadChannel: a small interface LeadMailbox now implements (publish/peek/ack
  plus a selfCoordId() accessor). It exists so FleetMcp and the receive loop can
  be tested with a fake instead of a live broker. LeadMailbox's AMQP logic is
  untouched — the diff is the implements clause, four @Override marks and the
  accessor.

- Fleetd.openLeadMailbox: opens this daemon's mailbox after the reply inbox is
  selected, with the same env-injected seam selectReplyInbox uses. Every "off"
  path returns null and the daemon still starts: no coordinator block (silent),
  a uriEnv that does not resolve (INFO), a configured broker with no selfId
  (WARN — a mailbox is named after the coord-id that owns it), or a broker that
  refuses at boot (WARN, credentials stripped). Closed in the ordered shutdown
  hook, after the loop that reads it has stopped.

- fleet_send{coordId}: publishes a LeadMessage(from=selfCoordId, to=coordId) to
  the peer's mailbox and returns the broker-confirmed receipt. coordId is
  mutually exclusive with sessionId/turnId and is rejected by name rather than
  resolved by precedence. An unroutable/nacked/timed-out publish comes back as a
  tool error naming the coordId, never a crash. The worker send/reply path is
  not touched.

- LeadCoordLoop: the receive half. Each tick peeks the mailbox, resolves the
  local lead pane, and — only at a turn boundary — injects "[lead <from>] <text>"
  and acks. Anything not delivered stays unacked and is retried, so a message is
  never dropped; one message per tick, so every delivery is gated on a status
  read that already saw the previous one.

- fleet_list reports {selfId, configured} when coordination is on, so an
  operator can find the coord-id a peer must use to reach them. Omitted
  entirely when it is off.

Tests: 20 new hermetic tests (no broker) across routing, delivery and startup
selection. mvn clean install: Tests run: 944, Failures: 0, Errors: 0, Skipped: 0
— BUILD SUCCESS.
2026-08-24 17:39:29 +02:00
Dai Ha f29968c334 Merge autocompact-window: per-profile autoCompactWindow for claude-code + opencode members 2026-08-24 17:32:53 +02:00
Dai Ha 2757bc7185 autoCompactWindow: per-profile bounded auto-compaction, both backends
CI / contract (pull_request) Successful in 41s
CI / build (pull_request) Failing after 1m37s
Add opt-in Integer autoCompactWindow to FleetConfig.Profile (last field,
null/unset = today's behaviour). Validated at config load to [100000,
1000000] — the band Claude Code's own --autocompact flag accepts.

Claude Code: appends --autocompact <window> to argv (mirrors --model),
so it survives the ccs <profile> wrapper.

opencode: has no absolute compact-at-N knob (only compaction.auto/prune/
reserved/tail_turns/preserve_recent_tokens), so the window is applied as
the resolved model's own limit.context (+ a required limit.output:16384
default) in the generated opencode.json, merged via get-or-create nodes
so it does not clobber a custom-provider block. Only applies when model:
resolves to "provider/model"; otherwise logs a WARN naming the profile
rather than silently doing nothing.

Docs added to fleetd.example.yaml explaining the cross-backend semantics
difference (compacts AT the window vs. WITHIN it).
2026-08-24 16:54:40 +02:00
18 changed files with 1293 additions and 15 deletions
+2 -1
View File
@@ -112,7 +112,8 @@ the merge — and merging on a reviewer's word is delegating it by proxy.
| Delegate (blocking) | `fleet_send{sessionId, content}` |
| Delegate (long task) | `fleet_send{sessionId, content, wait:false}` → ticket → `fleet_poll{ticket}` |
| Answer a member's `fleet_ask` | `fleet_send{turnId, content}` — **not** `sessionId` |
| Message a **peer lead** | `fleet_send{sessionId: <their terminal>, content}` — `fleet_list` → `leads` reports it. Coordination only, **never** a task |
| Message a **peer lead** on this host | `fleet_send{sessionId: <their terminal>, content}` — `fleet_list` → `leads` reports it. Coordination only, **never** a task |
| Message a **peer lead** on another daemon or host | `fleet_send{coordId: <their coord-id>, content}` — needs a `coordinator:` block; your own coord-id is in `fleet_list`. Coordination only, **never** a task |
| Answer a peer lead that messaged you | `fleet_reply{content}` — the one case a lead replies |
| Collect a held reply | `fleet_poll{target}` · then `fleet_ack{target, msgId}` |
| Tear down a member | `fleet_stop{paneId}` |
+13
View File
@@ -140,6 +140,18 @@ herdrSocket: ~/.config/herdr/herdr.sock
# e.g. `env DISPLAY=:10.0 idea {dir}`. Best-effort: a failure is logged, never
# fails the spawn. Omit to open the member's module by hand. There is no close
# half yet — an opened module stays open until the operator closes it.
# autoCompactWindow → opt-in, default off. A bounded token window that forces a spawned member to
# compact its context instead of running on the backend's own default and dying
# mid-turn (losing its fleet_reply — the whole point of the turn — with it).
# Validated at config load to [100000, 1000000] — the band Claude Code's own
# --autocompact flag accepts.
# CROSS-BACKEND SEMANTICS DIFFER: on claude-code this is a launch-time
# `--autocompact <tokens>` flag — the member compacts AT this window. opencode
# has no equivalent flag (it only forces `compaction.auto: true`, unconditionally,
# already), so this is instead applied as the model's `limit.context` in the
# generated opencode.json — the member compacts WITHIN this window, not exactly
# at it — and only when this profile's `model:` is in `provider/model` form; if it
# isn't, bridged logs a WARN naming the profile rather than silently doing nothing.
# tokenEnv → host env var holding the worker's auth token (value never stored in config);
# omit for a backend that needs no token (e.g. a local ollama).
# cwd → pin this profile's working directory (CB-112). Omit to inherit the primary's
@@ -256,6 +268,7 @@ profiles:
# ideMcpUrl: http://127.0.0.1:29170/index-mcp/streamable-http # opt-in (CB-634): IDE code intelligence, pinned to the worktree
# ideProjectDir: bridged # CB-634: module dir the IDE opens + the overlay pins (this repo's pom is in bridged/)
# ideOpenCommand: env DISPLAY=:10.0 idea {dir} # CB-634 auto-open: opens {dir} in the IDE at spawn; omit to open by hand
# autoCompactWindow: 250000 # opt-in: bound member context; claude-code compacts AT this, opencode within it (model limit.context)
gx11: # a second backend, so `placement: weighted` has a choice
baseUrl: http://gx01.gw:8000 # self-hosted; ccs handles the model + token
placement: tab
@@ -31,6 +31,9 @@ import dev.ltms.fleet.mcp.LsofPeerPidLookup;
import dev.ltms.fleet.mcp.LsofProcessCwdLookup;
import dev.ltms.fleet.msg.AmqpReplyInbox;
import dev.ltms.fleet.msg.InMemoryReplyInbox;
import dev.ltms.fleet.msg.LeadChannel;
import dev.ltms.fleet.msg.LeadCoordLoop;
import dev.ltms.fleet.msg.LeadMailbox;
import dev.ltms.fleet.msg.MessageService;
import dev.ltms.fleet.msg.Rendezvous;
import dev.ltms.fleet.msg.ReplyInbox;
@@ -60,6 +63,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -79,6 +83,14 @@ public final class Fleetd {
/** CB-504: how long to wait at startup for herdr's socket before serving degraded. */
private static final long HERDR_WAIT_SECONDS = 30;
/**
* CB-637: how often the lead coordination loop looks for peer messages. A few seconds — slow
* enough that an idle fleet is not polling a broker in a tight loop, fast enough that a peer
* lead's message is not left sitting once the local lead reaches a turn boundary. The mailbox
* pushes into the loop's held set on its own consumer thread, so this interval bounds only the
* pane delivery, never the receive.
*/
private static final long LEAD_COORD_INTERVAL_MS = 3_000L;
private static final long HERDR_WAIT_POLL_MILLIS = 500;
/**
@@ -375,6 +387,12 @@ public final class Fleetd {
// unusable), bridged stays soft-state on the in-memory inbox. The AMQP inbox owns a broker
// connection, so keep the reference to close it in the ordered shutdown hook.
final ReplyInbox replyInbox = selectReplyInbox(cfg.broker(), System.getenv(), AmqpReplyInbox::open);
// CB-637: this daemon's lead-to-lead mailbox on the SHARED coordination vhost — a separate
// broker from the reply inbox by design (see FleetConfig.Coordinator). Absent a coordinator:
// block this is null and every lead path below is simply not wired, which is exactly the
// behaviour before this ticket. It owns a broker connection, so keep the reference for the
// ordered shutdown hook.
final LeadMailbox leadMailbox = openLeadMailbox(cfg.coordinator(), System.getenv(), LeadMailbox::open);
// CB-307: learn the primary's terminal from orchestration tool calls (or pin from config).
// The pin also feeds CallerResolver below: a primary running inside a herdr pane would
// otherwise resolve as a worker and be refused every orchestration tool.
@@ -503,7 +521,27 @@ public final class Fleetd {
new FleetMcp.QuarantineSource(profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.effectiveCredentialId();
}, quarantine));
}, quarantine),
leadMailbox);
// CB-637: the receive half. Only constructed when a lead mailbox actually opened — with no
// coordinator (or an unreachable one) there is nothing to deliver, so no scheduler is
// created and no thread runs. It reads the SAME live lead supplier the injector's
// deliverability gate does, so a lead found by the tab scan after startup is reachable
// without a restart.
final LeadCoordLoop leadCoordLoop;
final ScheduledExecutorService leadCoordSchedulerRef;
if (leadMailbox != null) {
var leadCoordScheduler = Executors.newSingleThreadScheduledExecutor(r ->
Thread.ofVirtual().name("bridge-leadcoord-").unstarted(r));
leadCoordLoop = new LeadCoordLoop(leadMailbox, agents, leads, leadCoordScheduler,
LEAD_COORD_INTERVAL_MS);
leadCoordLoop.start();
leadCoordSchedulerRef = leadCoordScheduler;
} else {
leadCoordLoop = null;
leadCoordSchedulerRef = null;
}
// 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.
@@ -524,6 +562,8 @@ public final class Fleetd {
messages.close();
pushLoop.close();
if (heartbeat != null) heartbeat.close(); // CB-551: stop the idle-lead heartbeat scheduler
if (leadCoordLoop != null) leadCoordLoop.close(); // CB-637: stop delivering peer-lead messages
if (leadCoordSchedulerRef != null) leadCoordSchedulerRef.shutdownNow();
if (healthMonitor != null) healthMonitor.stop();
if (configWatcher != null) configWatcher.stop(); // CB-559: stop polling the config file
mcp.close();
@@ -536,6 +576,15 @@ public final class Fleetd {
log.debug("reply inbox close: {}", e.toString());
}
}
// CB-637: the coordination connection goes with it — after the loop that reads it has
// stopped, so no tick can be mid-ack against a closed channel.
if (leadMailbox != null) {
try {
leadMailbox.close();
} catch (Exception e) {
log.debug("lead mailbox close: {}", e.toString());
}
}
herdr.close();
}));
@@ -575,6 +624,66 @@ public final class Fleetd {
ReplyInbox open(String uri, int prefetch);
}
/** Injection seam for {@link #openLeadMailbox}: production binds {@link LeadMailbox#open}. */
@FunctionalInterface
interface LeadMailboxOpener {
LeadMailbox open(String uri, String selfCoordId, int prefetch);
}
/**
* CB-637: open this daemon's lead-to-lead mailbox, or return {@code null} to leave the feature
* off. Package-private and env-injected for the same reason as {@link #selectReplyInbox}: the
* selection is then testable without a broker or a mutable process environment.
*
* <p>Every "off" path returns {@code null}, and each says why at the level it deserves:
*
* <ul>
* <li>no {@code coordinator:} block — silent. Lead coordination is opt-in; an operator who
* never configured it does not need to be told it is off on every boot.</li>
* <li>a block whose {@code uriEnv} does not resolve — INFO, the same "you moved to the secret
* store and the variable is not there" case {@code selectReplyInbox} warns about.</li>
* <li>a configured broker but no {@code selfId} — WARN. This one is a half-finished config: a
* mailbox is named after the coord-id that owns it, so with no id there is no queue to own
* and no {@code from} to send as. Loud, because the operator plainly intended the feature.</li>
* <li>the broker refuses at boot — WARN, and carry on. Mirrors {@code openAmqpOrFallback}: a
* coordination broker that is down must never take a whole fleet's daemon with it, and the
* fleet still works exactly as it did before this feature existed.</li>
* </ul>
*/
static LeadMailbox openLeadMailbox(FleetConfig.Coordinator coordinator, Map<String, String> env,
LeadMailboxOpener opener) {
if (coordinator == null) {
return null; // opt-in: nothing configured, nothing to say
}
String uri = coordinator.effectiveUri(env);
if (uri == null) {
log.info("lead coordination: OFF — coordinator{} has no usable broker uri",
coordinator.uriEnv() == null ? "" : ".uriEnv=" + coordinator.uriEnv());
return null;
}
if (coordinator.selfId() == null || coordinator.selfId().isBlank()) {
log.warn("coordinator.selfId is unset — lead coordination is OFF. A lead mailbox is the "
+ "queue named after the coord-id that owns it, so with no id there is nothing to "
+ "own and no sender identity to publish as. Set coordinator.selfId to a name that "
+ "is unique across every daemon sharing {} and restart bridged.",
stripCredentials(uri));
return null;
}
try {
LeadMailbox mailbox = opener.open(uri, coordinator.selfId(), coordinator.prefetchOrDefault());
log.info("lead coordination: ON as coord-id {} (prefetch={})",
coordinator.selfId(), coordinator.prefetchOrDefault());
return mailbox;
} catch (IllegalStateException e) {
log.warn("cannot reach the AMQP coordination broker ({}) — lead-to-lead messaging is OFF "
+ "for this process lifetime. fleet_send{{coordId}} will report it as "
+ "not configured, and peer messages already queued stay on the broker until "
+ "a restart picks them up. Reason: {}",
stripCredentials(uri), reasonOf(e));
return null;
}
}
/**
* CB-151/152: pick the reply inbox. A usable broker — a literal {@code uri}, or a {@code
* uriEnv} whose variable resolves (both read from {@code env}) — selects the durable AMQP inbox.
@@ -294,6 +294,19 @@ public record FleetConfig(
* profile that does not opt in. Read live off the current config, so it is
* HOT: a change takes effect on the next exhaustion classification / spawn,
* no restart needed.
* @param autoCompactWindow opt-in per-profile token window that forces a spawned member to
* auto-compact its context at (Claude Code) or within (opencode) a bound the
* operator chooses, instead of the backend's own default. {@code null} (the
* default) leaves today's behaviour exactly — opencode already forces
* {@code compaction.auto: true} unconditionally (CB-523) but has no absolute
* window, and Claude Code has neither. When set, validated at config load to
* {@code [100000, 1000000]} — the band Claude Code's own {@code --autocompact
* <tokens>} flag accepts. The two backends honour it differently: Claude Code
* compacts AT this window (a launch-time {@code --autocompact} flag);
* opencode has no such knob, so this is applied as the model's
* {@code limit.context} instead, which bounds the window opencode compacts
* <em>within</em>, and only when {@code model} resolves to a
* {@code provider/model} pair.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
public record Profile(String profile, String baseUrl, String model,
@@ -311,7 +324,8 @@ public record FleetConfig(
String credentialId,
String ideMcpUrl,
String ideProjectDir,
String ideOpenCommand) {
String ideOpenCommand,
Integer autoCompactWindow) {
/** Peer kind spawned by {@link dev.ltms.fleet.member.ClaudeCodeLauncher} (the default). */
public static final String KIND_CLAUDE_CODE = "claude-code";
@@ -381,6 +395,10 @@ public record FleetConfig(
// substituted; blank ⇒ no auto-open (the operator opens the module by hand).
ideProjectDir = (ideProjectDir == null || ideProjectDir.isBlank()) ? null : ideProjectDir;
ideOpenCommand = (ideOpenCommand == null || ideOpenCommand.isBlank()) ? null : ideOpenCommand;
// Opt-in per profile, default off (null). No clamping here — unlike ideMcpUrl/ideProjectDir
// there is no blank-string form to normalize (it's an Integer), and the [100000, 1000000]
// range is enforced eagerly at config load (rejectAutoCompactWindowOutOfRange), naming the
// profile, rather than silently clamped here. A profile that never sets it keeps null.
}
/**
@@ -425,7 +443,29 @@ public record FleetConfig(
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, credentialId, ideMcpUrl, ideProjectDir, ideOpenCommand);
exhaustedPattern, credentialId, ideMcpUrl, ideProjectDir, ideOpenCommand, autoCompactWindow);
}
/**
* Backward-compatible constructor without the {@code autoCompactWindow} field — the profile
* leaves auto-compaction at the backend's own default (opencode's unconditional
* {@code compaction.auto: true}, or Claude Code's built-in threshold), exactly as before this
* key existed. This is the shape the canonical constructor had before the field was added —
* every pre-existing Java call site (and any YAML that omits the key) keeps compiling and
* behaving identically; Jackson binds the canonical (longest) constructor, so YAML omitting
* {@code autoCompactWindow:} still lands here as {@code null} via that path, not this one.
*/
public Profile(String profile, String baseUrl, String model,
String configDir, String tokenEnv, List<String> argv,
String placement, String workspace, String tabLabel, String mcpUrl,
String cwd, List<String> parityOverlay, String gitTokenEnv, String gitHostEnv,
String kind, Map<String, String> env, Float weight, Integer maxLoad,
Boolean subscription, String exhaustedPattern, String credentialId,
String ideMcpUrl, String ideProjectDir, String ideOpenCommand) {
this(profile, baseUrl, model, configDir, tokenEnv, argv, placement, workspace, tabLabel,
mcpUrl, cwd, parityOverlay, gitTokenEnv, gitHostEnv, kind, env, weight, maxLoad,
subscription, exhaustedPattern, credentialId, ideMcpUrl, ideProjectDir, ideOpenCommand,
null);
}
/**
@@ -1290,6 +1330,7 @@ public record FleetConfig(
warnUnknownTopLevelKeys(yaml, path);
rejectDuplicateMemberSlots(yaml);
rejectNegativeMaxLoad(yaml);
rejectAutoCompactWindowOutOfRange(yaml);
rejectUnknownKind(yaml);
rejectUnknownAuthMode(yaml);
rejectUnknownPlacement(yaml);
@@ -1592,6 +1633,52 @@ public record FleetConfig(
}
}
/** Lowest {@code autoCompactWindow} Claude Code's {@code --autocompact <tokens>} flag accepts. */
static final int AUTO_COMPACT_WINDOW_MIN = 100_000;
/** Highest {@code autoCompactWindow} Claude Code's {@code --autocompact <tokens>} flag accepts. */
static final int AUTO_COMPACT_WINDOW_MAX = 1_000_000;
/**
* Reject a profile whose {@code autoCompactWindow:} is set but outside the token band Claude
* Code's own {@code --autocompact <tokens>} flag accepts (100k–1M), naming both the profile and
* the value.
*
* <p>Unset/{@code null} means "off" and passes silently — today's behaviour for every profile
* that does not opt in (see {@link Profile#autoCompactWindow()}). A profile that DOES set the key
* is validated eagerly, at config load, rather than failing later when Claude Code itself refuses
* the launch flag on spawn — the same "fail loud at load, not lazily at first spawn" reasoning as
* {@link #rejectNegativeMaxLoad} and {@link #rejectUnknownPlacementPolicy}.
*
* @param yaml the raw config text
* @throws IllegalStateException when any profile's {@code autoCompactWindow} is set and outside
* {@code [100000, 1000000]}
*/
static void rejectAutoCompactWindowOutOfRange(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("profiles") instanceof Map<?, ?> profiles)) {
return;
}
List<String> bad = profiles.entrySet().stream()
.filter(e -> e.getValue() instanceof Map<?, ?> p
&& p.get("autoCompactWindow") instanceof Number n
&& (n.doubleValue() < AUTO_COMPACT_WINDOW_MIN || n.doubleValue() > AUTO_COMPACT_WINDOW_MAX))
.map(e -> String.valueOf(e.getKey()))
.sorted()
.toList();
if (!bad.isEmpty()) {
throw new IllegalStateException("refusing to start: profile(s) [" + String.join(", ", bad)
+ "] set autoCompactWindow outside [" + AUTO_COMPACT_WINDOW_MIN + ", "
+ AUTO_COMPACT_WINDOW_MAX + "] — Claude Code's --autocompact flag accepts only "
+ "that band of tokens; omit the key to leave auto-compaction at each backend's "
+ "own default.");
}
}
/** The peer kinds this build has an adapter for — {@link Profile#kind()}'s only valid values. */
private static final Set<String> KNOWN_KINDS = Set.of(Profile.KIND_CLAUDE_CODE, Profile.KIND_OPENCODE);
@@ -11,6 +11,8 @@ import dev.ltms.fleet.metrics.FleetMetrics;
import dev.ltms.fleet.metrics.Metrics;
import dev.ltms.fleet.inject.MemberPresence;
import dev.ltms.fleet.herdr.HerdrException;
import dev.ltms.fleet.msg.LeadChannel;
import dev.ltms.fleet.msg.LeadMessage;
import dev.ltms.fleet.msg.MessageService;
import dev.ltms.fleet.msg.Rendezvous;
import dev.ltms.fleet.peer.PeerUnreachableException;
@@ -37,6 +39,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import java.util.function.Function;
@@ -97,6 +100,8 @@ public final class FleetMcp {
private final CapacitySource capacity;
private final HealthCoverageSource healthCoverage;
private final QuarantineSource quarantine;
/** CB-637: this daemon's lead-to-lead channel; {@code null} when no coordinator is configured. */
private final LeadChannel leadChannel;
/** Capacity facts used by {@code fleet_list}; production must supply the placement live count. */
public record CapacitySource(Function<String, Integer> liveCount, Function<String, Integer> maxLoad,
@@ -131,6 +136,21 @@ public final class FleetMcp {
ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage,
QuarantineSource quarantine) {
this(messages, workers, sessions, identity, presence, primaryRegistry, callers, metrics, capacity,
healthCoverage, quarantine, null);
}
/**
* As above, with this daemon's lead-to-lead channel (CB-637). {@code leadChannel} is
* {@code null} whenever no {@code coordinator:} block is configured or its broker could not be
* reached at boot — cross-daemon lead messaging is simply off, and {@code fleet_send{coordId}}
* says so rather than failing obscurely.
*/
public FleetMcp(MessageService messages, PeerLauncher workers, SessionManager sessions,
ConnectionIdentity identity, MemberPresence presence, PrimaryRegistry primaryRegistry,
CallerResolver callers, Metrics metrics, CapacitySource capacity, HealthCoverageSource healthCoverage,
QuarantineSource quarantine, LeadChannel leadChannel) {
this.leadChannel = leadChannel;
this.capacity = capacity;
this.quarantine = Objects.requireNonNull(quarantine, "quarantine");
this.healthCoverage = healthCoverage;
@@ -177,6 +197,13 @@ public final class FleetMcp {
String target = str(a, "sessionId");
String content = str(a, "content");
String turnId = str(a, "turnId");
String coordId = str(a, "coordId");
if (coordId != null && !coordId.isBlank()) {
// CB-637: a peer LEAD on another daemon, addressed by coord-id over the shared
// coordination broker. Checked before the turnId branch so a call that sets both
// is rejected as the conflict it is, rather than silently taking one route.
return sendToLead(leadChannel, coordId, content, target, turnId);
}
if (turnId != null && !turnId.isBlank()) {
// Answering a worker's fleet_ask (CB-205): resolve its blocked question and
// block for the worker's reply as it resumes the same turn. This is the same
@@ -257,7 +284,8 @@ public final class FleetMcp {
if (denied != null) return denied;
return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine,
callers == null ? Map.of() : callers.leads(),
callerTerminal(exchange));
callerTerminal(exchange),
leadChannel == null ? null : leadChannel.selfCoordId());
};
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> stopHandler =
(exchange, req) -> {
@@ -585,6 +613,54 @@ public final class FleetMcp {
return null;
}
/**
* {@code fleet_send} carrying a {@code coordId} (CB-637): a message to a PEER LEAD, published to
* that lead's durable mailbox on the shared coordination broker. This is the only lead→lead path
* that crosses hosts — the existing pane-injection route can only reach a lead whose herdr socket
* this daemon shares.
*
* <p>{@code coordId} is mutually exclusive with {@code sessionId} and {@code turnId}: those two
* address a worker session owned by <em>this</em> daemon, a coord-id addresses a lead owned by
* another one, and there is no sensible reading of a call that sets both. Rejected by name rather
* than resolved by precedence, so a caller that meant the other route learns it instead of having
* its message quietly go somewhere else.
*
* <p>The publish is synchronous and confirmed by the broker, so the result is a real delivery
* receipt rather than a hopeful one. Its failure — nobody owns {@code coordId}'s mailbox, the
* broker nacked, or the confirm timed out — arrives as {@link IllegalStateException} and is
* turned into a tool error naming the coord-id. It is never allowed to escape as a crash: an
* unreachable peer is an ordinary outcome of addressing a fleet you do not control.
*
* @param leadChannel this daemon's channel, or {@code null} when no coordinator is configured
*/
static McpSchema.CallToolResult sendToLead(LeadChannel leadChannel, String coordId, String content,
String sessionId, String turnId) {
if (!isBlank(sessionId) || !isBlank(turnId)) {
String conflict = !isBlank(sessionId) ? "sessionId" : "turnId";
return error("coordId and " + conflict + " are mutually exclusive: coordId addresses a peer "
+ "LEAD on another daemon over the coordination broker, while " + conflict
+ " addresses a worker session on this one. Pass exactly one.");
}
if (isBlank(content)) {
return error("content is required");
}
if (leadChannel == null) {
return error("lead coordination is not configured (no coordinator: block) — cannot send to "
+ "peer lead \"" + coordId + "\". Add a coordinator: block with a shared broker uri "
+ "and this daemon's selfId, then restart bridged.");
}
LeadMessage msg = new LeadMessage(UUID.randomUUID().toString(), leadChannel.selfCoordId(),
coordId, content);
try {
leadChannel.publish(coordId, msg);
} catch (IllegalStateException e) {
return error("cannot deliver to peer lead \"" + coordId + "\": " + e.getMessage()
+ ". Check that a daemon is running with coordinator.selfId=\"" + coordId
+ "\" and is connected to the same coordination broker.");
}
return text("delivered to peer lead " + coordId + " (msgId " + msg.msgId() + ")");
}
/** {@code fleet_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)) {
@@ -870,6 +946,22 @@ public final class FleetMcp {
static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
CapacitySource capacity, HealthCoverageSource healthCoverage,
QuarantineSource quarantine, Map<String, String> leads, String selfTerm) {
return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, leads, selfTerm, null);
}
/**
* As above, additionally reporting this daemon's own lead coordination id (CB-637) when one is
* configured and its channel opened. There is no peer-discovery surface yet — a lead addresses a
* peer by a coord-id it was told — so this row exists to answer the one question the operator
* cannot answer any other way: what is MY coord-id, the one a peer must use to reach me. It is
* omitted entirely when no coordinator is configured, so an ordinary fleet's output is unchanged.
*
* @param selfCoordId this daemon's coord-id, or {@code null} when lead coordination is off
*/
static McpSchema.CallToolResult listFleet(PeerLauncher workers, SessionManager sessions, MessageService messages,
CapacitySource capacity, HealthCoverageSource healthCoverage,
QuarantineSource quarantine, Map<String, String> leads, String selfTerm,
String selfCoordId) {
try {
Map<String, Agent> live = workers.list().stream()
.map(Agent.class::cast)
@@ -888,6 +980,9 @@ public final class FleetMcp {
Map<String, Object> result = new LinkedHashMap<>();
result.put("leads", leadRows); result.put("members", out);
result.put("healthCoverage", healthCoverage.value().get());
if (selfCoordId != null && !selfCoordId.isBlank()) {
result.put("coordinator", Map.of("selfId", selfCoordId, "configured", true));
}
if (capacity.available()) result.put("capacity", profiles.stream()
.map(profile -> capacityView(profile, capacity.liveCount(), capacity.maxLoad(), roster, messages,
capacity.clock().getAsLong(), quarantine)).toList());
@@ -1015,7 +1110,9 @@ public final class FleetMcp {
"Delegate a task to a worker session. By default blocks until the worker replies and "
+ "returns its reply (or a 'still working / queued' note on timeout). Pass wait:false "
+ "for a long task to return a ticket immediately, then poll it with fleet_poll. To "
+ "answer a worker's fleet_ask, pass its turnId (with content) instead of sessionId.",
+ "answer a worker's fleet_ask, pass its turnId (with content) instead of sessionId. "
+ "To message a PEER LEAD on another daemon — possibly another host — pass its "
+ "coordId instead; that is coordination, never a task.",
objectSchema(Map.of(
"sessionId", stringProp("The worker session id (herdr terminal_id) to delegate to"),
"content", stringProp("The task/message to send to the worker (or your answer, with turnId)"),
@@ -1023,7 +1120,11 @@ public final class FleetMcp {
"wait", Map.of("type", "boolean",
"description", "Block for the reply (default true); false returns a ticket to poll"),
"turnId", stringProp("When answering a worker's fleet_ask, its question turnId — "
+ "routes your answer back into the same turn (omit for a normal delegation)")),
+ "routes your answer back into the same turn (omit for a normal delegation)"),
"coordId", stringProp("A peer LEAD's coordination id — delivers content to that "
+ "lead's durable mailbox on the shared coordination broker, which works "
+ "across hosts. Mutually exclusive with sessionId and turnId. Your own "
+ "coordId is reported by fleet_list.")),
List.of("content")));
}
@@ -248,7 +248,7 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
// has neither MCP nor a charter — session flags must be added into a list we own.
List<String> argv = mutableArgv(argvWithFleet(cfg, spec));
String agentSessionId = applySessionIdentity(argv, spec.sessionName(), spec.resumeSessionId());
return new Launch(workerEnv, argvWithModel(argv, cfg), agentSessionId);
return new Launch(workerEnv, argvWithAutoCompact(argvWithModel(argv, cfg), cfg), agentSessionId);
}
/**
@@ -463,6 +463,30 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
return withModel;
}
/**
* Pin a bounded auto-compaction window on the command line via {@code --autocompact <tokens>},
* opt-in per profile (CB-634's sibling ticket: a member that runs out of context dies mid-turn
* and its {@code fleet_reply} — the whole point of the turn — is lost with it; opencode already
* forces {@code compaction.auto: true} unconditionally, CB-523, but Claude Code has no equivalent
* and runs at the backend's own default window).
*
* <p>Mirrors {@link #argvWithModel}: appended after it, so it survives the {@code ccs <profile>}
* wrapper the same way {@code --model} does, and outranks env/settings and the operator's own
* {@code argv}. Verified: {@code claude 2.1.241 --help} lists {@code --autocompact <auto|tokens>}
* (either the literal {@code auto}, or an integer 100k–1M) — {@link FleetConfig#load} rejects a
* configured value outside that band before this ever runs, so the flag Claude Code receives here
* is always in range.
*/
private static List<String> argvWithAutoCompact(List<String> argv, FleetConfig.Profile cfg) {
if (cfg.autoCompactWindow() == null) {
return argv;
}
List<String> withAutoCompact = mutableArgv(argv);
withAutoCompact.add("--autocompact");
withAutoCompact.add(String.valueOf(cfg.autoCompactWindow()));
return withAutoCompact;
}
// --- Agent-returning convenience spawns (used by callers/tests that want the herdr Agent) ---
/** Spawn a worker for the default profile in the resolved default cwd. */
@@ -209,9 +209,19 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
@Override
protected Launch buildLaunch(FleetConfig.Profile cfg, LaunchSpec spec) {
Map<String, String> workerEnv = baseEnv(cfg);
// autoCompactWindow's opencode lever (limit.context) only targets a specific provider/model
// entry, so it needs model: in "provider/model" form. A profile that opts in without that
// shape gets no silent no-op — log it, once, here, whether or not writeConfig ends up running.
boolean wantsContextLimit = cfg.autoCompactWindow() != null && splitProviderModel(cfg.model()) != null;
if (cfg.autoCompactWindow() != null && !wantsContextLimit) {
log.warn("profile '{}' sets autoCompactWindow but model '{}' is not \"<provider>/<model>\" "
+ "form — opencode's per-model context limit could not be applied for this profile",
cfg.profile(), cfg.model());
}
// A config file is needed for the bridge MCP mount, a member charter, the IDE MCP (+ its
// guidance overlay, CB-634), or a pinned endpoint (CB-508).
if (cfg.hasMcp() || cfg.hasIdeMcp() || spec.charter() != null || hasCustomProvider(cfg)) {
// guidance overlay, CB-634), a pinned endpoint (CB-508), or a resolvable autoCompactWindow.
if (cfg.hasMcp() || cfg.hasIdeMcp() || spec.charter() != null || hasCustomProvider(cfg)
|| wantsContextLimit) {
workerEnv.put("OPENCODE_CONFIG", writeConfig(cfg, spec.charter(), spec.cwd()).toString());
}
applyGitToken(workerEnv, cfg);
@@ -360,6 +370,9 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
if (hasCustomProvider(cfg)) {
addCustomProvider(root, cfg);
}
if (cfg.autoCompactWindow() != null) {
applyContextLimit(root, cfg);
}
Path cfgFile = dir.resolve("opencode.json");
// Built with Jackson rather than string concatenation: the provider block is nested and
@@ -404,18 +417,68 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
* default gateway — a worker quietly talking to the wrong endpoint is the failure this avoids.
*/
private static String[] splitModelSelector(FleetConfig.Profile cfg) {
String model = cfg.model();
int slash = model == null ? -1 : model.indexOf('/');
if (model == null || model.isBlank() || slash <= 0 || slash == model.length() - 1) {
String[] parts = splitProviderModel(cfg.model());
if (parts == null) {
throw new IllegalArgumentException(
"profile " + cfg.profile() + " sets baseUrl (a pinned opencode endpoint) so"
+ " model: must be \"<provider>/<model>\", e.g."
+ " \"local-vllm/deepseek-v4-flash\"; got "
+ (model == null ? "null" : '"' + model + '"'));
+ (cfg.model() == null ? "null" : '"' + cfg.model() + '"'));
}
return parts;
}
/**
* Split {@code model} into its {@code provider} and {@code model} halves, or {@code null} when
* it is not in that shape (unset/blank, or no non-trailing {@code /}). Unlike
* {@link #splitModelSelector}, non-throwing — callers that only *optionally* need the split
* (autoCompactWindow's context-limit application) use this to fall back to a WARN rather than an
* exception, since a profile without {@code baseUrl} is not required to name a provider/model.
*/
private static String[] splitProviderModel(String model) {
int slash = model == null ? -1 : model.indexOf('/');
if (model == null || model.isBlank() || slash <= 0 || slash == model.length() - 1) {
return null;
}
return new String[]{model.substring(0, slash), model.substring(slash + 1)};
}
/**
* Apply the per-profile {@code autoCompactWindow} as opencode's per-model context limit.
*
* <p>opencode has no absolute "compact at N tokens" knob — its {@code compaction} block only
* exposes {@code auto}/{@code prune}/{@code reserved}/{@code tail_turns}/
* {@code preserve_recent_tokens} — so the real lever is the model's own
* {@code provider.<p>.models.<m>.limit.context}, which bounds the window opencode compacts
* <em>within</em> rather than compacting exactly AT it the way Claude Code's {@code --autocompact}
* does.
*
* <p>Uses get-or-create nodes ({@code withObject}) at every level so this MERGES with any provider
* block {@link #addCustomProvider} already wrote for a custom-provider (pinned-endpoint) profile —
* it must never overwrite that block's {@code npm}/{@code name}/{@code options}. For a gateway
* profile (no {@code baseUrl}, so no prior provider block) this writes a partial
* {@code provider.<p>.models.<m>.limit} override, which opencode merges over its own built-in
* provider definition.
*
* <p>opencode's {@code limit} schema requires both {@code context} and {@code output}; there is no
* independent signal for the latter here, so 16384 is written as a safe default (documented in
* {@code fleetd.example.yaml}).
*
* <p>Silently does nothing when {@code model:} is not in {@code provider/model} form — a warning
* for that case is already logged once in {@code buildLaunch}, so this stays quiet rather than
* duplicating it.
*/
private static void applyContextLimit(ObjectNode root, FleetConfig.Profile cfg) {
String[] parts = splitProviderModel(cfg.model());
if (parts == null) {
return;
}
ObjectNode limit = root.withObject("provider").withObject(parts[0])
.withObject("models").withObject(parts[1]).withObject("limit");
limit.put("context", cfg.autoCompactWindow());
limit.put("output", 16384);
}
/**
* The OpenAI-compatible base URL for {@code baseUrl}. A bare {@code host:port} gets {@code /v1}
* appended (where these servers put the API); a URL that already carries a path is taken as-is,
@@ -0,0 +1,40 @@
package dev.ltms.fleet.msg;
import java.util.List;
/**
* The lead-to-lead message channel this daemon speaks, as its callers need it — one lead's own
* mailbox: publish to a peer's coord-id, look at what has arrived for me, and ack what I have
* delivered.
*
* <p>Extracted from {@link LeadMailbox} purely as a seam. {@code LeadMailbox} is the one production
* implementation and owns a live AMQP connection, so a test that wanted to exercise the routing in
* {@code FleetMcp} or the delivery in {@link LeadCoordLoop} would have had to stand up a broker —
* which is exactly the kind of test that gets tagged {@code contract} and then does not run. With
* this interface both of those are hermetic: they inject a fake channel and assert on what was
* published, peeked and acked.
*
* <p>Note what is <em>not</em> here: {@code drain()} and {@code close()}. Draining is a convenience
* over peek+ack that no caller on this seam uses, and closing is the owner's job — {@code Fleetd}
* holds the concrete {@link LeadMailbox} for its shutdown hook and hands only this narrower view to
* everyone else.
*/
public interface LeadChannel {
/**
* Send {@code m} to {@code toCoordId}'s mailbox, blocking until the broker confirms it is
* durably queued. Throws {@link IllegalStateException} when it is not — unroutable (nobody owns
* that coord-id), nacked, or unconfirmed within the implementation's timeout. A caller must
* report that as a failed send, never as a delivered one.
*/
void publish(String toCoordId, LeadMessage m);
/** Non-destructive FIFO snapshot of the messages held for this daemon's own coord-id. */
List<LeadMessage> peek();
/** Drop {@code msgId} from the held set and ack it on the broker. A no-op if it is not held. */
void ack(String msgId);
/** This daemon's own lead coordination id — the mailbox it owns, and the {@code from} it sends as. */
String selfCoordId();
}
@@ -0,0 +1,193 @@
package dev.ltms.fleet.msg;
import dev.ltms.fleet.herdr.AgentControl;
import dev.ltms.fleet.herdr.AgentStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
/**
* The receive half of lead-to-lead messaging: a bounded background loop that takes what has arrived
* in this daemon's own {@link LeadChannel} mailbox and types it into the local lead's herdr pane.
*
* <p>{@code fleet_send{coordId}} is the send half — it publishes to a peer daemon's mailbox and
* returns. Nothing on the receiving side reads that mailbox on its own, because the peer lead is an
* interactive agent, not a service that polls; this loop is what closes the gap.
*
* <p><strong>Status-gated, exactly like {@link ReplyPushLoop}.</strong> A pane may only be injected
* into at a turn boundary ({@link AgentStatus#injectable()} — idle, blocked or done); pasting into
* a live turn corrupts it. So a tick that finds the lead busy simply does nothing and comes back
* later.
*
* <p><strong>Ack only after delivery.</strong> A message is acked — removed from the broker — only
* once {@link AgentControl#send} has actually put it in the pane. Anything not delivered (no lead
* pane resolvable, lead mid-turn, herdr threw) stays unacked and is retried on the next tick, and
* survives a daemon restart because the broker still holds it. The cost of that choice is a
* possible duplicate — the send lands and the ack does not — which is the right way round: a peer
* lead seeing a message twice is a nuisance, a peer lead never seeing it at all is the failure this
* whole path exists to remove.
*
* <p><strong>One message per tick.</strong> The loop delivers at most one held message per tick even
* when several are waiting. Injecting a second one immediately would mean acting on a status read
* taken <em>before</em> the first injection: that first paste starts a turn, and herdr does not
* report the pane as {@code working} the instant it does. Waiting for the next tick means every
* delivery is gated on a status read that already saw the previous one. A backlog therefore drains
* one message per {@code intervalMs}, in FIFO order.
*/
public final class LeadCoordLoop {
private static final Logger log = LoggerFactory.getLogger(LeadCoordLoop.class);
/** How an arriving peer message is rendered into the lead's pane — the sender's coord-id, then its text. */
static final String DELIVERY_FORMAT = "[lead %s] %s";
private final LeadChannel channel;
private final AgentControl agents;
private final Supplier<Map<String, String>> leads;
private final ScheduledExecutorService scheduler;
private final long intervalMs;
private volatile boolean running;
/**
* @param channel this daemon's own lead mailbox
* @param agents herdr control, for the status gate and the pane injection
* @param leads live {@code terminal_id → name} view of the leads this daemon recognises —
* read through the supplier on every tick, never snapshotted, so a lead found by
* the tab scan after startup becomes reachable without a restart
* @param scheduler the loop's own scheduler; the caller owns its shutdown
* @param intervalMs how long between ticks
*/
public LeadCoordLoop(LeadChannel channel, AgentControl agents, Supplier<Map<String, String>> leads,
ScheduledExecutorService scheduler, long intervalMs) {
this.channel = channel;
this.agents = agents;
this.leads = leads;
this.scheduler = scheduler;
this.intervalMs = intervalMs;
}
/** Begin ticking. Idempotent-ish: calling it twice would schedule two chains, so call it once. */
public void start() {
running = true;
log.info("lead coordination: delivering peer messages for coord-id {} every {}ms",
channel.selfCoordId(), intervalMs);
scheduleNext();
}
/** Stop ticking. In-flight work finishes; nothing further is scheduled. */
public void close() {
running = false;
}
private void scheduleNext() {
if (!running) {
return;
}
scheduler.schedule(this::tickAndReschedule, intervalMs, TimeUnit.MILLISECONDS);
}
private void tickAndReschedule() {
try {
tick();
} catch (RuntimeException e) {
// Never let one bad tick end the chain — the next one re-reads everything from scratch.
log.warn("lead coordination tick failed: {}", e.toString());
}
scheduleNext();
}
/**
* One tick: deliver at most one held peer message into the local lead's pane and ack it.
* Package-private so a test drives it directly rather than waiting on the scheduler.
*/
void tick() {
List<LeadMessage> held;
try {
held = channel.peek();
} catch (RuntimeException e) {
log.debug("lead coordination: cannot read the mailbox this tick: {}", e.toString());
return;
}
if (held.isEmpty()) {
return;
}
String lead = resolveLocalLead();
if (lead == null) {
// Left unacked on purpose: the broker keeps holding it until a lead pane exists.
log.debug("lead coordination: {} message(s) waiting but no local lead pane to deliver to",
held.size());
return;
}
AgentStatus status;
try {
status = agents.status(lead);
} catch (RuntimeException e) {
log.debug("lead coordination: status check failed for lead {}, retrying next tick", lead, e);
return;
}
if (!status.injectable()) {
log.debug("lead coordination: lead {} is {} (not injectable), holding {} message(s)",
lead, status, held.size());
return;
}
LeadMessage msg = held.getFirst();
try {
agents.send(lead, DELIVERY_FORMAT.formatted(msg.from(), msg.content()));
} catch (RuntimeException e) {
// Not delivered, so not acked — the broker still has it for the next tick.
log.warn("lead coordination: failed to deliver message {} from {} to lead {}: {}",
msg.msgId(), msg.from(), lead, e.toString());
return;
}
try {
channel.ack(msg.msgId());
} catch (RuntimeException e) {
// Delivered but not acked: it will be redelivered, which the javadoc calls out as the
// deliberate direction of this trade.
log.warn("lead coordination: delivered message {} but could not ack it: {}",
msg.msgId(), e.toString());
return;
}
log.debug("lead coordination: delivered message {} from {} to lead {}", msg.msgId(), msg.from(), lead);
}
/**
* Which local pane a peer's message is for. The mailbox's {@code selfCoordId} is this daemon's
* one lead identity, so there is exactly one right answer — this only has to find it:
*
* <ol>
* <li>a lead whose configured name equals {@code selfCoordId} — the explicit, unambiguous case;</li>
* <li>otherwise the sole lead, when this daemon recognises exactly one;</li>
* <li>otherwise nothing, and the message waits.</li>
* </ol>
*
* <p>Step 3 is deliberate rather than a guess-the-lead fallback. Picking one of several leads
* arbitrarily would type a peer's message into a pane it was not addressed to, and the message
* would then be acked and gone. Leaving it held costs a delay and nothing else.
*/
private String resolveLocalLead() {
Map<String, String> known = leads.get();
if (known.isEmpty()) {
return null;
}
String self = channel.selfCoordId();
for (var entry : known.entrySet()) {
if (entry.getValue() != null && entry.getValue().equals(self)) {
return entry.getKey();
}
}
if (known.size() == 1) {
return known.keySet().iterator().next();
}
log.warn("lead coordination: {} leads are known and none is named \"{}\" — cannot tell which "
+ "pane a peer message is for; name one lead after coordinator.selfId to fix this",
known.size(), self);
return null;
}
}
@@ -61,7 +61,7 @@ import java.util.concurrent.TimeoutException;
* redelivery) and any publish still awaiting its confirm is failed rather than left to idle out
* the confirm timeout against a sequence number that means nothing on the new channel.
*/
public final class LeadMailbox implements AutoCloseable {
public final class LeadMailbox implements LeadChannel, AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(LeadMailbox.class);
@@ -195,6 +195,7 @@ public final class LeadMailbox implements AutoCloseable {
* confirmed within {@link #CONFIRM_TIMEOUT_MS} — the caller must treat that as a failed publish,
* not a lost-and-forgotten one.
*/
@Override
public void publish(String toCoordId, LeadMessage msg) {
byte[] body;
try {
@@ -239,7 +240,14 @@ public final class LeadMailbox implements AutoCloseable {
}
}
/** The coord-id whose mailbox this instance owns — the {@code from} of everything it publishes. */
@Override
public String selfCoordId() {
return selfCoordId;
}
/** Non-destructive FIFO snapshot of this mailbox's currently-held messages. */
@Override
public List<LeadMessage> peek() {
synchronized (held) {
return held.values().stream().map(Held::message).toList();
@@ -254,6 +262,7 @@ public final class LeadMailbox implements AutoCloseable {
}
/** Remove the held message {@code msgId} and ack it on the broker. No-op if not held. */
@Override
public void ack(String msgId) {
Held h;
synchronized (held) {
@@ -0,0 +1,144 @@
package dev.ltms.fleet;
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 dev.ltms.fleet.config.FleetConfig;
import dev.ltms.fleet.msg.LeadMailbox;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.*;
/**
* CB-637: the daemon decides whether lead-to-lead messaging is on in
* {@link Fleetd#openLeadMailbox}, not in the config record — so testing
* {@code Coordinator.isConfigured()} alone would pass even if {@code Fleetd} never honoured it.
* These drive the real selection with an injected env map and an injected opener, so no broker is
* involved and no process environment is mutated.
*
* <p>The invariant every case shares: the feature turns itself OFF, never takes the daemon down.
* A fleet whose coordination broker is missing, half-configured or unreachable must still start and
* still work exactly as it did before this feature existed.
*/
class FleetdLeadMailboxSelectionTest {
private static final String SECRET = "c00rdPw";
private static final String RESOLVED_URI = "amqp://user:" + SECRET + "@coord.example:5672/coord";
/** Fake opener: records what it was offered, or fails as an unreachable broker would. */
private static final class RecordingOpener implements Fleetd.LeadMailboxOpener {
String offeredUri;
String offeredSelfId;
int offeredPrefetch = -1;
boolean unreachable;
@Override
public LeadMailbox open(String uri, String selfCoordId, int prefetch) {
this.offeredUri = uri;
this.offeredSelfId = selfCoordId;
this.offeredPrefetch = prefetch;
if (unreachable) {
throw new IllegalStateException("cannot connect to AMQP coordination broker at " + uri,
new java.net.ConnectException("Connection refused"));
}
// A real LeadMailbox needs a live connection; nothing here dereferences the result
// beyond a null check, so the "reachable" cases assert on what was OFFERED instead.
return null;
}
}
private static ListAppender<ILoggingEvent> captureFleetdLogs() {
Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
return appender;
}
private static String joined(ListAppender<ILoggingEvent> appender, Level level) {
return appender.list.stream().filter(e -> e.getLevel() == level)
.map(ILoggingEvent::getFormattedMessage).reduce("", (a, b) -> a + "\n" + b);
}
@Test
void noCoordinatorBlockLeavesTheFeatureOffSilently() {
var appender = captureFleetdLogs();
var opener = new RecordingOpener();
assertNull(Fleetd.openLeadMailbox(null, Map.of(), opener));
assertNull(opener.offeredUri, "nothing configured means nothing is opened");
assertEquals("", joined(appender, Level.WARN),
"an opt-in feature nobody asked for must not warn on every boot");
}
@Test
void opensTheMailboxWhenAUriAndSelfIdAreConfigured() {
var opener = new RecordingOpener();
var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null);
Fleetd.openLeadMailbox(coordinator, Map.of(), opener);
assertEquals(RESOLVED_URI, opener.offeredUri);
assertEquals("mac-opus", opener.offeredSelfId);
assertEquals(LeadMailbox.DEFAULT_PREFETCH, opener.offeredPrefetch,
"an unset prefetch takes the mailbox's own default, not zero");
}
@Test
void honoursUriEnvOverALiteralUri() {
var opener = new RecordingOpener();
var coordinator = new FleetConfig.Coordinator("amqp://stale:stale@old:5672/x", "COORD_URI",
"mac-opus", 8);
Fleetd.openLeadMailbox(coordinator, Map.of("COORD_URI", RESOLVED_URI), opener);
assertEquals(RESOLVED_URI, opener.offeredUri, "the secret store wins over clear text");
assertEquals(8, opener.offeredPrefetch);
}
@Test
void turnsOffWhenUriEnvDoesNotResolve() {
var opener = new RecordingOpener();
var coordinator = new FleetConfig.Coordinator(null, "COORD_URI", "mac-opus", null);
assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener));
assertNull(opener.offeredUri);
}
@Test
void warnsAndStaysOffWhenSelfIdIsMissing() {
var appender = captureFleetdLogs();
var opener = new RecordingOpener();
var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, null, null);
assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener));
assertNull(opener.offeredUri, "a mailbox with no owning coord-id has no queue to declare");
String warns = joined(appender, Level.WARN);
assertTrue(warns.contains("coordinator.selfId"), () -> "say which key is missing: " + warns);
assertFalse(warns.contains(SECRET), () -> "the URI's password must never be logged: " + warns);
}
@Test
void warnsAndStaysOffWhenTheBrokerIsUnreachableAtBoot() {
var appender = captureFleetdLogs();
var opener = new RecordingOpener();
opener.unreachable = true;
var coordinator = new FleetConfig.Coordinator(RESOLVED_URI, null, "mac-opus", null);
assertNull(Fleetd.openLeadMailbox(coordinator, Map.of(), opener),
"a down coordination broker turns the feature off; it must never take the daemon down");
String warns = joined(appender, Level.WARN);
assertTrue(warns.contains("coord.example"), () -> "name the host that failed: " + warns);
assertFalse(warns.contains(SECRET), () -> "with credentials stripped: " + warns);
assertTrue(warns.contains("Connection refused"), () -> "and the real reason: " + warns);
}
}
@@ -43,6 +43,49 @@ class FleetConfigTest {
assertTrue(cfg.guard().hostSet().contains("ollama.ltms.dev"));
}
@Test
void aProfileWithAnAutoCompactWindowBelowTheAcceptedRangeIsRejectedAtLoad(@TempDir Path dir)
throws Exception {
Path f = dir.resolve("low-window.yaml");
Files.writeString(f, """
profiles:
ltms-local:
baseUrl: http://gx00.gw:8000
autoCompactWindow: 50000
""");
IllegalStateException e = assertThrows(IllegalStateException.class, () -> FleetConfig.load(f));
assertTrue(e.getMessage().contains("ltms-local"), "the offending profile is named");
assertTrue(e.getMessage().contains("autoCompactWindow"));
}
@Test
void aProfileWithAnAutoCompactWindowInRangeLoadsFine(@TempDir Path dir) throws Exception {
Path f = dir.resolve("in-range-window.yaml");
Files.writeString(f, """
profiles:
ltms-local:
baseUrl: http://gx00.gw:8000
autoCompactWindow: 250000
""");
FleetConfig cfg = FleetConfig.load(f);
assertEquals(250_000, cfg.profiles().get("ltms-local").autoCompactWindow());
}
@Test
void aProfileWithNoAutoCompactWindowLeavesItNull(@TempDir Path dir) throws Exception {
Path f = dir.resolve("no-window.yaml");
Files.writeString(f, """
profiles:
ltms-local:
baseUrl: http://gx00.gw:8000
""");
assertNull(FleetConfig.load(f).profiles().get("ltms-local").autoCompactWindow(),
"unset means off — today's behaviour, unchanged");
}
@Test
void appliesDefaultsForMissingSections(@TempDir Path dir) throws Exception {
Path f = dir.resolve("minimal.yaml");
@@ -0,0 +1,96 @@
package dev.ltms.fleet.mcp;
import dev.ltms.fleet.msg.FakeLeadChannel;
import dev.ltms.fleet.msg.LeadMessage;
import io.modelcontextprotocol.spec.McpSchema;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
/**
* CB-637: the {@code fleet_send{coordId}} route — a message to a PEER LEAD on another daemon,
* published to its durable mailbox instead of typed into a pane this daemon can reach.
*
* <p>Hermetic: {@link FakeLeadChannel} replaces the AMQP-backed {@code LeadMailbox}, so these run
* with no broker. What is checked here is routing and refusal — which envelope goes out, and which
* calls are rejected before anything is sent.
*/
class FleetMcpLeadCoordTest {
private static final String SELF = "mac-opus";
private static final String PEER = "fleet01-lead";
private static String textOf(McpSchema.CallToolResult r) {
return ((McpSchema.TextContent) r.content().getFirst()).text();
}
@Test
void publishesAnEnvelopeAddressedFromThisDaemonToThePeer() {
var channel = new FakeLeadChannel(SELF);
McpSchema.CallToolResult res =
FleetMcp.sendToLead(channel, PEER, "you own the auth layer, I own the config one", null, null);
assertFalse(res.isError(), () -> "expected a success result, got: " + textOf(res));
assertEquals(1, channel.published().size());
LeadMessage sent = channel.published().getFirst();
assertEquals(SELF, sent.from(), "the sender is this daemon's own coord-id, never an argument");
assertEquals(PEER, sent.to());
assertEquals("you own the auth layer, I own the config one", sent.content());
assertNotNull(sent.msgId());
assertFalse(sent.msgId().isBlank(), "the envelope carries an idempotency id for dedup on redelivery");
assertTrue(textOf(res).contains(PEER), "the receipt names the peer it reached");
}
@Test
void rejectsCoordIdTogetherWithSessionId() {
var channel = new FakeLeadChannel(SELF);
McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", "term_worker", null);
assertTrue(res.isError());
assertTrue(textOf(res).contains("coordId"), () -> textOf(res));
assertTrue(textOf(res).contains("sessionId"), () -> "the error must name the conflict: " + textOf(res));
assertEquals(0, channel.published().size(), "an ambiguous call must send nothing at all");
}
@Test
void rejectsCoordIdTogetherWithTurnId() {
var channel = new FakeLeadChannel(SELF);
McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", null, "turn_7");
assertTrue(res.isError());
assertTrue(textOf(res).contains("turnId"), () -> textOf(res));
assertEquals(0, channel.published().size());
}
@Test
void saysSoPlainlyWhenLeadCoordinationIsNotConfigured() {
McpSchema.CallToolResult res = FleetMcp.sendToLead(null, PEER, "hi", null, null);
assertTrue(res.isError());
assertTrue(textOf(res).contains("lead coordination is not configured"), () -> textOf(res));
assertTrue(textOf(res).contains("coordinator"), () -> "point at the config block to add: " + textOf(res));
}
@Test
void reportsAnUnreachablePeerAsAToolErrorNamingIt() {
var channel = new FakeLeadChannel(SELF)
.failPublishWith("lead message m1 was returned as unroutable (mailbox not owned)");
McpSchema.CallToolResult res = FleetMcp.sendToLead(channel, PEER, "hi", null, null);
assertTrue(res.isError(), "a black-holed message must never be reported as delivered");
assertTrue(textOf(res).contains(PEER), () -> "the error must name the coordId: " + textOf(res));
assertTrue(textOf(res).contains("unroutable"), () -> "and carry the broker's reason: " + textOf(res));
}
@Test
void requiresContent() {
var channel = new FakeLeadChannel(SELF);
assertTrue(FleetMcp.sendToLead(channel, PEER, " ", null, null).isError());
assertEquals(0, channel.published().size());
}
}
@@ -527,6 +527,35 @@ class FleetMcpTest {
assertTrue(out.contains("\"liveStatus\":\"unknown\""), out);
}
@Test
void listReportsThisDaemonsOwnCoordIdWhenLeadCoordinationIsOn() {
FakeHerdr h = new FakeHerdr();
SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
McpSchema.CallToolResult res = FleetMcp.listFleet(
workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, null,
FleetMcp.CapacitySource.none(), new FleetMcp.HealthCoverageSource(() -> "off"),
FleetMcp.QuarantineSource.none(), Map.of(), "", "mac-opus");
String out = textOf(res);
// There is no peer-discovery surface yet, so this row answers the one question an operator
// cannot answer any other way: which coord-id a peer must use to reach ME.
assertTrue(out.contains("\"coordinator\""), out);
assertTrue(out.contains("\"selfId\":\"mac-opus\""), out);
}
@Test
void listOmitsTheCoordinatorRowWhenLeadCoordinationIsOff() {
FakeHerdr h = new FakeHerdr();
SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
McpSchema.CallToolResult res = FleetMcp.listFleet(
workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), sessions, Map.of(), "");
assertFalse(textOf(res).contains("coordinator"),
"an ordinary fleet's output must be unchanged by this feature");
}
@Test
void capacityUsesThePlacementLiveCount() {
FakeHerdr h = new FakeHerdr();
@@ -1138,6 +1138,42 @@ class ClaudeCodeLauncherTest {
assertNull(startEnv(herdr).get("ANTHROPIC_MODEL"));
}
// ── autoCompactWindow: --autocompact is pinned on the command line, opt-in per profile ───────
/** A launcher for a profile identical but for its {@code autoCompactWindow:} — the only variable. */
private ClaudeCodeLauncher serviceWithAutoCompactWindow(FakeHerdr herdr, Integer window) {
FleetConfig.Profile cfg = profileWithAutoCompactWindow(window);
return new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
_ -> null);
}
private static FleetConfig.Profile profileWithAutoCompactWindow(Integer window) {
return new FleetConfig.Profile("sonnet", "http://gx00.gw:8000", "claude-sonnet-5", null,
"BRIDGED_WORKER_TOKEN", List.of("ccs", "sonnet"), "tab", "bridged-workers",
"w #{n}", "http://127.0.0.1:8765/mcp", null, null, null, null, null, Map.of(),
null, null, null, null, null, null, null, null, window);
}
@Test
void aConfiguredAutoCompactWindowIsPassedAsAnAutocompactFlag() {
FakeHerdr herdr = new FakeHerdr();
serviceWithAutoCompactWindow(herdr, 250_000).spawn("sonnet", null, null);
List<String> args = spawnedArgs(herdr);
int flag = args.indexOf("--autocompact");
assertTrue(flag >= 0, "the flag is what survives a wrapper argv like [ccs, sonnet]");
assertEquals("250000", args.get(flag + 1));
}
@Test
void aProfileWithNoAutoCompactWindowGetsNoAutocompactFlag() {
FakeHerdr herdr = new FakeHerdr();
serviceWithAutoCompactWindow(herdr, null).spawn("sonnet", null, null);
assertFalse(spawnedArgs(herdr).contains("--autocompact"));
}
// --- CB-539: subscription-profile opt-in ----------------------------------------------------
/** A claude-code profile on the subscription: no baseUrl (by design), no off-sub endpoint. */
@@ -495,6 +495,71 @@ class OpenCodeLauncherTest {
"without a baseUrl opencode resolves its own provider as before");
}
// --- autoCompactWindow: opencode has no absolute compact-at-N knob, so this is applied as the
// model's own limit.context, only when model: resolves to "provider/model" -----------------
private static FleetConfig.Profile opencodeCfgWithAutoCompactWindow(String model, String baseUrl,
Integer window) {
return new FleetConfig.Profile("gemini", baseUrl, model, null, "BRIDGED_WORKER_TOKEN",
List.of("opencode"), "tab", "bridged-workers", "opencode: {model} #{n}", null,
null, null, null, null, FleetConfig.Profile.KIND_OPENCODE, Map.of(), null, null,
null, null, null, null, null, null, window);
}
@Test
void autoCompactWindowIsAppliedAsThePerModelContextLimit(@TempDir Path root) throws Exception {
FakeHerdr herdr = new FakeHerdr();
service(herdr, root, opencodeCfgWithAutoCompactWindow("openai/gpt-5", null, 250_000)).spawn();
String cfgPath = startEnv(herdr).get("OPENCODE_CONFIG");
assertNotNull(cfgPath, "autoCompactWindow alone must trigger config generation, with no MCP"
+ " and no custom provider set");
JsonNode limit = new ObjectMapper().readTree(Path.of(cfgPath).toFile())
.path("provider").path("openai").path("models").path("gpt-5").path("limit");
assertEquals(250_000, limit.path("context").asInt());
assertEquals(16384, limit.path("output").asInt(),
"opencode's limit schema requires both keys; output gets a safe documented default");
}
@Test
void autoCompactWindowMergesIntoACustomProviderRatherThanOverwritingIt(@TempDir Path root)
throws Exception {
FakeHerdr herdr = new FakeHerdr();
service(herdr, root, opencodeCfgWithAutoCompactWindow(
"local-vllm/deepseek-v4-flash", "http://127.0.0.1:8000", 300_000)).spawn();
JsonNode provider = new ObjectMapper()
.readTree(Path.of(startEnv(herdr).get("OPENCODE_CONFIG")).toFile())
.path("provider").path("local-vllm");
assertEquals("@ai-sdk/openai-compatible", provider.path("npm").asText(),
"addCustomProvider's own fields must survive the later limit merge");
assertEquals(300_000, provider.path("models").path("deepseek-v4-flash")
.path("limit").path("context").asInt());
assertEquals("deepseek-v4-flash", provider.path("models").path("deepseek-v4-flash")
.path("name").asText(),
"the model's pre-existing 'name' field must survive the limit merge too");
}
@Test
void autoCompactWindowWithNoProviderSlashInModelGetsNoLimitAndAWarn(@TempDir Path root)
throws Exception {
FakeHerdr herdr = new FakeHerdr();
// mcpUrl set too, only so a config file gets written at all to inspect; a bare model name
// with no other config-triggering knob would leave OPENCODE_CONFIG unset entirely, which is
// also correct (nothing to write) but not what this test is asserting.
service(herdr, root, new FleetConfig.Profile("gemini", null, "some-free-model", null,
"BRIDGED_WORKER_TOKEN", List.of("opencode"), "tab", "bridged-workers",
"opencode: {model} #{n}", "http://127.0.0.1:8765/mcp", null, null, null, null,
FleetConfig.Profile.KIND_OPENCODE, Map.of(), null, null, null, null, null, null,
null, null, 250_000)).spawn();
String cfgPath = startEnv(herdr).get("OPENCODE_CONFIG");
JsonNode json = new ObjectMapper().readTree(Path.of(cfgPath).toFile());
assertTrue(json.path("provider").isMissingNode(),
"a bare model name cannot be targeted at a specific provider/model limit entry — "
+ "no silent no-op, but also no broken partial write");
}
// --- CB-634: IDE Index MCP + guidance overlay (opencode does not read CLAUDE.local.md) -------
private static FleetConfig.Profile opencodeIdeCfg(String mcpUrl, String ideUrl, String cwd) {
@@ -0,0 +1,75 @@
package dev.ltms.fleet.msg;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
/**
* Hermetic stand-in for {@link LeadChannel}: an in-memory mailbox that records what was published
* and what was acked, with no broker anywhere.
*
* <p>Its whole point is that {@link LeadMailbox} — the production implementation — owns a live AMQP
* connection, so every test of the code AROUND it would otherwise need a broker and end up tagged
* {@code contract}. The broker round trip is covered once, by {@code LeadMailboxTest}; the routing
* ({@code FleetMcp}) and the delivery ({@link LeadCoordLoop}) are covered here.
*
* <p>Thread-safe: {@link LeadCoordLoop} calls it from its own scheduler thread while a test reads
* the recorded lists.
*/
public final class FakeLeadChannel implements LeadChannel {
private final String selfCoordId;
private final List<LeadMessage> held = Collections.synchronizedList(new ArrayList<>());
private final List<LeadMessage> published = Collections.synchronizedList(new ArrayList<>());
private final List<String> acked = Collections.synchronizedList(new ArrayList<>());
/** When set, every {@link #publish} throws it — the unroutable/nacked/timed-out peer. */
private volatile IllegalStateException publishFailure;
public FakeLeadChannel(String selfCoordId) {
this.selfCoordId = selfCoordId;
}
/** Make every publish fail as an unreachable peer would. */
public FakeLeadChannel failPublishWith(String message) {
this.publishFailure = new IllegalStateException(message);
return this;
}
/** Put a message in this mailbox as if a peer had sent it. */
public FakeLeadChannel hold(LeadMessage m) {
held.add(m);
return this;
}
@Override
public void publish(String toCoordId, LeadMessage m) {
if (publishFailure != null) {
throw publishFailure;
}
published.add(m);
}
@Override
public List<LeadMessage> peek() {
return List.copyOf(held);
}
@Override
public void ack(String msgId) {
held.removeIf(m -> m.msgId().equals(msgId));
acked.add(msgId);
}
@Override
public String selfCoordId() {
return selfCoordId;
}
public List<LeadMessage> published() {
return List.copyOf(published);
}
public List<String> acked() {
return List.copyOf(acked);
}
}
@@ -0,0 +1,150 @@
package dev.ltms.fleet.msg;
import dev.ltms.fleet.herdr.AgentControl;
import dev.ltms.fleet.herdr.FakeHerdr;
import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import static org.junit.jupiter.api.Assertions.*;
/**
* Unit tests for {@link LeadCoordLoop} — the receive half of lead-to-lead messaging. Hermetic
* throughout: a {@link FakeLeadChannel} stands in for the mailbox and {@link FakeHerdr} for the
* pane, so no broker and no herdr daemon is involved. The broker round trip has its own
* {@code contract}-tagged test on {@link LeadMailbox}.
*
* <p>Every test drives {@link LeadCoordLoop#tick()} directly rather than waiting on the scheduler:
* the schedule itself is one {@code scheduler.schedule} call, while the decisions worth pinning —
* inject or hold, ack or leave unacked — all live in the tick.
*/
class LeadCoordLoopTest {
private static final String SELF = "mac-opus";
private static final String PEER = "fleet01-lead";
private static final String LEAD_TERM = "term_lead";
/** No scheduler is needed: nothing here calls start(). */
private static final ScheduledExecutorService NO_SCHEDULER = null;
private static LeadCoordLoop loop(LeadChannel channel, FakeHerdr herdr, Map<String, String> leads) {
return new LeadCoordLoop(channel, new AgentControl(herdr), () -> leads, NO_SCHEDULER, 3_000L);
}
private static List<FakeHerdr.Call> prompts(FakeHerdr herdr) {
return herdr.calls.stream().filter(c -> c.method().equals("agent.prompt")).toList();
}
@Test
void deliversAHeldMessageToTheLeadPaneAndAcksIt() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "the merge is blocked"));
var herdr = new FakeHerdr().agentStatus("idle");
loop(channel, herdr, Map.of(LEAD_TERM, SELF)).tick();
assertEquals(1, prompts(herdr).size(), "an injectable lead must receive the peer's message");
@SuppressWarnings("unchecked")
Map<String, Object> params = (Map<String, Object>) prompts(herdr).getFirst().params();
assertEquals(LEAD_TERM, params.get("target"), "delivered to the resolved local lead pane");
assertEquals("[lead " + PEER + "] the merge is blocked", params.get("text"),
"the sender's coord-id is carried into the pane — the lead must know who to answer");
assertEquals(List.of("m1"), channel.acked(), "a delivered message is acked off the broker");
assertTrue(channel.peek().isEmpty(), "and is no longer held");
}
@Test
void leavesTheMessageUnackedWhenTheLeadIsMidTurn() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
var herdr = new FakeHerdr().agentStatus("working");
loop(channel, herdr, Map.of(LEAD_TERM, SELF)).tick();
assertEquals(0, prompts(herdr).size(), "never paste into a live turn");
assertEquals(List.of(), channel.acked(), "an undelivered message must NOT be acked");
assertEquals(1, channel.peek().size(), "it stays held for the next tick");
}
@Test
void leavesTheMessageUnackedWhenNoLeadPaneIsKnown() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
var herdr = new FakeHerdr().agentStatus("idle");
loop(channel, herdr, Map.of()).tick();
assertEquals(0, prompts(herdr).size());
assertEquals(List.of(), channel.acked(), "nowhere to deliver is not a reason to drop it");
assertEquals(1, channel.peek().size());
}
@Test
void leavesTheMessageUnackedWhenHerdrRefusesTheInjection() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
var herdr = new FakeHerdr().agentStatus("idle").agentSendFailsWith("agent_not_found");
loop(channel, herdr, Map.of(LEAD_TERM, SELF)).tick();
assertEquals(List.of(), channel.acked(), "a send that threw delivered nothing, so nothing is acked");
assertEquals(1, channel.peek().size());
}
@Test
void resolvesTheLeadByNameWhenSeveralAreKnown() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
var herdr = new FakeHerdr().agentStatus("idle");
// Two leads on this daemon; only one carries the coord-id the mailbox is owned as.
var leads = new java.util.LinkedHashMap<String, String>();
leads.put("term_other", "some-other-lead");
leads.put(LEAD_TERM, SELF);
loop(channel, herdr, leads).tick();
@SuppressWarnings("unchecked")
Map<String, Object> params = (Map<String, Object>) prompts(herdr).getFirst().params();
assertEquals(LEAD_TERM, params.get("target"), "the lead named after coordinator.selfId wins");
}
@Test
void holdsWhenSeveralLeadsAreKnownAndNoneCarriesTheCoordId() {
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
var herdr = new FakeHerdr().agentStatus("idle");
var leads = new java.util.LinkedHashMap<String, String>();
leads.put("term_one", "lead-one");
leads.put("term_two", "lead-two");
loop(channel, herdr, leads).tick();
assertEquals(0, prompts(herdr).size(),
"guessing a pane would type a peer's message into the wrong lead and then ack it");
assertEquals(List.of(), channel.acked());
}
@Test
void deliversOneMessagePerTickSoEachIsGatedOnItsOwnStatusRead() {
var channel = new FakeLeadChannel(SELF)
.hold(new LeadMessage("m1", PEER, SELF, "first"))
.hold(new LeadMessage("m2", PEER, SELF, "second"));
var herdr = new FakeHerdr().agentStatus("idle");
var loop = loop(channel, herdr, Map.of(LEAD_TERM, SELF));
loop.tick();
assertEquals(List.of("m1"), channel.acked(), "FIFO: the oldest goes first");
assertEquals(1, prompts(herdr).size(), "the second waits for a fresh status read");
loop.tick();
assertEquals(List.of("m1", "m2"), channel.acked());
assertEquals(2, prompts(herdr).size());
assertTrue(channel.peek().isEmpty());
}
@Test
void anEmptyMailboxNeverTouchesHerdr() {
var channel = new FakeLeadChannel(SELF);
var herdr = new FakeHerdr().agentStatus("idle");
loop(channel, herdr, Map.of(LEAD_TERM, SELF)).tick();
assertEquals(0, herdr.calls.size(), "an idle fleet must not poll a pane's status every tick");
}
}