Compare commits

..

8 Commits

Author SHA1 Message Date
Dai Ha bbbb4c1eb3 fleetd #302: require content in MessageService.reply so a REST reply with no content cannot silently resolve a waiter
CI / build (pull_request) Successful in 1m44s
CI / contract (pull_request) Successful in 3m17s
2026-09-04 12:31:09 +07:00
Dai Ha 38c248e617 Merge #298: a released reply is requeued, not dropped
CI / contract (push) Successful in 1m27s
CI / build (push) Successful in 1m46s
release() cancelled the consumer and dropped its local record of deliveries the
broker still held as outstanding. Cancelling a consumer does not requeue them:
they stay unacked on the still-open channel until it or the connection closes.
So a held-but-undrained reply became permanently unreachable — a worker's
report lost with no error and no log line. It now nacks with requeue, after
cancelling, so a later own() can still receive it.
2026-09-04 12:21:30 +07:00
Dai Ha 9debc0de27 #297: render both profiles doors from one body builder, not two copies
CI / contract (push) Successful in 1m23s
CI / build (push) Successful in 1m31s
The merged change shared the QuarantineSource and OutageSource instances
between fleet_profiles and GET /profiles, so the two doors read identical
facts. It then rendered those facts through a character-for-character copy of
the loop, in a different file. Shared inputs do not make duplicated computation
safe: a later edit to the row shape lands on one door and not the other, and
the two disagree about a live outage. That is what #284 was.

The ticket caused this. It said 'read from the same shared instances' and 'do
not change FleetMcp', and together those made copying the loop the only legal
move. Extracting FleetMcp.profilesView and calling it from both is what the
ticket should have asked for.
2026-09-04 12:19:02 +07:00
Dai Ha f34361b263 Merge #297: REST reports what MCP reports
GET /agents and GET /members now map a herdr transport failure into the same
{error, detail} envelope every other handler in FleetApp uses, instead of
letting it escape to Javalin's default handling. GET /profiles now reports the
quarantined and coolingOff states, read from the same shared sources FleetMcp
reads. REST is the door a lead falls back to when its MCP mount drops, so it
was weakest exactly when it was load-bearing.
2026-09-04 12:16:00 +07:00
Dai Ha 85c90d440a #297: map HerdrException on GET /agents and /members; GET /profiles reports quarantine + cool-off
CI / contract (pull_request) Successful in 45s
CI / build (pull_request) Successful in 1m33s
Two REST-only visibility gaps, both against the same shared instances FleetMcp
reads (BackendQuarantine/BackendOutagePolicy), never recomputed:

- GET /agents and GET /members let a HerdrException escape uncaught, outside
  the {error, detail} envelope every other failure path in FleetApp uses.
  Both now route through the existing herdrError() helper, matching healthz/
  sessionStatus. GET /members is the endpoint's own comment names as the
  out-of-band path a lead falls back to when its MCP mount drops.
- GET /profiles omitted the two outage states fleet_profiles already reports:
  quarantined (CB-578 stage B) and coolingOff (fleetd #201 Unit 5). FleetApp
  now takes the SAME FleetMcp.QuarantineSource/OutageSource instances Fleetd
  wires into FleetMcp (extracted to local vars in Fleetd.java so both doors
  share one object, not two independently-built copies of the same rule).

FleetMcp itself is unchanged. Item 3 of the ticket (a capacity block on
GET /members) is explicitly out of scope and was not added.
2026-09-04 12:12:35 +07:00
Dai Ha e97502d550 drainAll: record that the timeout is a whole-drain budget, not a per-session grace
CI / build (push) Successful in 1m27s
CI / contract (push) Successful in 1m48s
The javadoc said 'for each session that is BUSY, poll up to timeoutNanos',
which reads as a per-session grace period. The deadline is taken once, before
the loop, so the first BUSY session can spend all of it. That is deliberate and
is the safer of the two designs: the drain is one phase of a shutdown sequence
that must finish inside launchd's exit window, and a per-session grace would
overrun it and get the daemon SIGKILLed part-way through, leaving the sessions
not yet reached with no clean release, no preserved-worktree log and no
snapshot. Found by a read-only hunt that read the code correctly and drew the
opposite conclusion from the wording.
2026-09-04 12:09:44 +07:00
Dai Ha aef14ff46e Merge #296: a failed spawn no longer leaks the pane it opened
Two exits created a pane and left it running. The readiness gate propagated an
unrelated herdr error without teardown, and spawnAsPane never closed the pane it
split when the peer failed to start. Neither could be cleaned up by the caller:
SessionManager.acquire never learns the pane id, because spawn throws before it
returns one. It removed the worktree anyway, so the leak was a live backend with
a deleted cwd, invisible to fleet_list and holding a seat nothing decremented.
2026-09-04 12:08:23 +07:00
Dai Ha cba516bda4 fleetd #296: close panes on failed spawn
CI / contract (pull_request) Successful in 1m38s
CI / build (pull_request) Successful in 2m7s
2026-09-04 12:05:26 +07:00
9 changed files with 360 additions and 53 deletions
@@ -625,6 +625,19 @@ public final class Fleetd {
log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
}
// fleetd #297: named once and reused verbatim below for FleetApp's GET /profiles, rather than
// built a second time — two independently-constructed sources reading the SAME BackendQuarantine
// / BackendOutagePolicy would still be able to drift (e.g. a future edit to the credentialIdFor
// closure in only one of the two places), exactly the shape #284 was.
FleetMcp.QuarantineSource quarantineSource = new FleetMcp.QuarantineSource(profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.effectiveCredentialId();
}, quarantine);
FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.effectiveCredentialId();
}, outagePolicy);
FleetMcp mcp = new FleetMcp(messages, workers, sessions, identity, presence,
primaryRegistry, callers, metrics, new FleetMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
profile -> {
@@ -636,15 +649,9 @@ public final class Fleetd {
return FleetHealthMonitor.coverage(health != null && health.isEnabled(),
health != null && health.notifications() != null && health.notifications().configured());
}),
new FleetMcp.QuarantineSource(profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.effectiveCredentialId();
}, quarantine),
quarantineSource,
leadMailbox,
new FleetMcp.OutageSource(profile -> {
var configured = config.get().profiles().get(profile);
return configured == null ? null : configured.effectiveCredentialId();
}, outagePolicy),
outageSource,
new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)));
// CB-637: the receive half. Only constructed when a lead mailbox actually opened — with no
@@ -715,9 +722,12 @@ public final class Fleetd {
// GET /sessions must merge across both, or a down/unpolled member daemon is invisible.
// fleetd #111: live (re-read-per-request) memberCredentials view for GET /member-credentials —
// same hot-reload shape as the memberCredentials supplier passed to ClaudeCodeLauncher above.
// fleetd #297: quarantineSource/outageSource are the SAME instances passed to FleetMcp above —
// GET /profiles must report the identical quarantine/cool-off facts as fleet_profiles.
Javalin app = new FleetApp(herdr, memberHerdr, workers, sessions, messages, presence, mcp.servlet(),
callers, metrics, deliverable,
() -> MemberCredentialPolicyView.of(config.get().memberCredentials())).build();
() -> MemberCredentialPolicyView.of(config.get().memberCredentials()),
quarantineSource, outageSource).build();
app.start(cfg.bind().host(), cfg.bind().port());
log.info("fleetd listening on {}:{}, herdr socket {}",
cfg.bind().host(), cfg.bind().port(), socket);
@@ -1015,6 +1015,25 @@ public final class FleetMcp {
* both maps at once when it is both exhaustion-quarantined AND cooling off.
*/
static McpSchema.CallToolResult profiles(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
return text(json(profilesView(workers, quarantine, outage)));
}
/**
* The body both front doors answer {@code profiles} with: the configured profile names, the
* default, and the two independent outage states — {@code quarantined} (the backend reported it
* out of capacity) and {@code coolingOff} (the credential threw repeated non-exhaustion backend
* errors). Each map is present only when at least one profile is in that state, and a profile
* can appear in both at once, because the two checks are separate.
*
* <p>fleetd #297: extracted so {@code fleet_profiles} and {@code GET /profiles} render from ONE
* body builder rather than two copies. Passing both doors the same {@link QuarantineSource} and
* {@link OutageSource} instances is necessary but not sufficient: with the loop written out
* twice, a later edit to the row shape — a renamed key, an added field — lands on one door and
* not the other, and the two then disagree about a live outage. That is exactly what fleetd
* #284 was, where one rule computed in two places was widened in only one and a single response
* contradicted itself. Shared inputs do not make duplicated computation safe.
*/
public static Map<String, Object> profilesView(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
Map<String, Object> result = new LinkedHashMap<>();
result.put("profiles", workers.profiles());
result.put("default", workers.defaultProfile() == null ? "" : workers.defaultProfile());
@@ -1046,7 +1065,7 @@ public final class FleetMcp {
if (!coolingOff.isEmpty()) {
result.put("coolingOff", coolingOff);
}
return text(json(result));
return result;
}
/**
@@ -697,7 +697,20 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
if (paneId == null) {
throw new IllegalStateException("pane.split returned no pane — cannot start a peer");
}
Agent peer = startUniquelyNamed(cfg, argv, paneId).agent();
Agent peer;
try {
peer = startUniquelyNamed(cfg, argv, paneId).agent();
} catch (RuntimeException e) {
// The peer never started — don't leave the pane we just created orphaned.
// Best-effort cleanup; never let it mask the real spawn failure.
try {
stop(paneId);
} catch (RuntimeException cleanup) {
log.warn("failed to close orphaned pane {} after spawn error: {}",
paneId, cleanup.getMessage());
}
throw e;
}
log.info("{} started pane={} terminal={}", namePrefix, peer.paneId(), peer.terminalId());
return peer;
}
@@ -996,7 +1009,8 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
* that gap: it stops waiting immediately (never burns the rest of the timeout), runs the same
* teardown the timeout path below runs, and throws with a message that says the backend exited
* rather than that the pane was slow. Any other {@link HerdrException} still propagates
* unchanged — this gate does not know how to recover from it.
* unchanged — this gate does not interpret or recover from it, but it still closes the pane
* it opened before handing the exception to its caller.
*/
private void waitUntilInjectableOrThrow(String paneId) {
long start = nowMillis.getAsLong();
@@ -1010,7 +1024,15 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
if (isAlreadyGone(e)) {
failFastOnGoneBackend(paneId, e, nowMillis.getAsLong() - start);
}
throw e; // any other herdr failure is not ours to interpret — let it propagate
// This gate must not interpret an unrelated herdr error, but the caller does not
// receive paneId when spawn throws. Close the pane here before propagating e unchanged.
try {
stop(paneId);
} catch (RuntimeException cleanup) {
log.warn("failed to close orphaned pane {} after readiness-gate error: {}",
paneId, cleanup.getMessage());
}
throw e;
}
lastStatus = sample.status();
if (lastStatus.injectable() || refinedInjectable(paneId, sample)) {
@@ -408,10 +408,28 @@ public final class MessageService {
* as the zero-candidate case does, and let {@link #abandon} apply the eventual recovery
* deterministically instead.
*
* <p><strong>{@code content} is required (fleetd #302).</strong> Both doors that reach this
* method must reject a missing/blank reply the same way, so the check lives here rather than in
* either caller: {@code FleetMcp.reply} already refuses a {@code null} content before it ever
* calls this method (its own required-arg guard), and no test or production call site anywhere
* in the codebase relies on replying with empty content — confirmed by searching every call site
* of this method before adding the check, not assumed. Without this guard, a REST {@code
* POST /sessions/{id}/reply} whose body omits {@code content} (or a client library that maps a
* missing field to {@code ""}) used to reach {@link Rendezvous#resolve} with an empty string,
* silently completing the lead's blocking wait with nothing — indistinguishable from a worker
* that genuinely replied with nothing, which is worse than a loud failure because it destroys the
* information that the reply never arrived.
*
* @throws IllegalArgumentException if {@code content} is {@code null} or blank — the caller must
* report this as a client error (REST: 400 {@code bad_request}) rather than resolve
* anything
* @return always {@code true} — the reply resolved a live send, completed a parked ticket, or
* was queued
*/
public boolean reply(String session, String content) {
if (content == null || content.isBlank()) {
throw new IllegalArgumentException("content is required");
}
if (rendezvous.resolve(session, content)) {
count(FleetMetrics.REPLIES, "path", "rendezvous");
return true; // a live send took it — unchanged fast path
@@ -9,6 +9,7 @@ import dev.ltms.fleet.auth.Principal;
import dev.ltms.fleet.guard.GuardException;
import dev.ltms.fleet.metrics.FleetMetrics;
import dev.ltms.fleet.metrics.Metrics;
import dev.ltms.fleet.mcp.FleetMcp;
import dev.ltms.fleet.herdr.Agent;
import dev.ltms.fleet.herdr.HerdrClient;
import dev.ltms.fleet.herdr.HerdrException;
@@ -86,6 +87,12 @@ public final class FleetApp {
// absent() (the honest "no policy configured" view) for every constructor that does not wire
// a real one, so existing legacy call sites keep building without knowing this field exists.
private final Supplier<MemberCredentialPolicyView> memberCredentials;
// fleetd #297: the SAME shared sources FleetMcp.profiles/fleet_profiles reads (BackendQuarantine
// and BackendOutagePolicy are each one instance for the whole daemon — see Fleetd wiring) so
// GET /profiles cannot drift from fleet_profiles about which profile is quarantined/cooling off.
// .none() (the honest "feature not wired" view) for every constructor that does not pass one.
private final FleetMcp.QuarantineSource quarantine;
private final FleetMcp.OutageSource outage;
private final ObjectMapper mapper = new ObjectMapper();
/**
@@ -146,6 +153,23 @@ public final class FleetApp {
MessageService messages, MemberPresence presence,
HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials) {
this(herdr, memberHerdr, workers, sessions, messages, presence, mcpServlet, auth, metrics,
deliverable, memberCredentials, FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
}
/**
* @param quarantine the SAME {@link FleetMcp.QuarantineSource} instance passed to {@code
* FleetMcp} (fleetd #297), so {@code GET /profiles} reports the identical
* exhaustion-quarantine facts as {@code fleet_profiles} rather than a second,
* independently-computed copy
* @param outage the SAME {@link FleetMcp.OutageSource} instance passed to {@code FleetMcp} —
* see {@code quarantine}; a SEPARATE check from it, never merged in
*/
public FleetApp(HerdrClient herdr, HerdrClient memberHerdr, PeerLauncher workers, SessionManager sessions,
MessageService messages, MemberPresence presence,
HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials,
FleetMcp.QuarantineSource quarantine, FleetMcp.OutageSource outage) {
this.herdr = herdr;
this.memberHerdr = memberHerdr != null ? memberHerdr : herdr;
this.workers = workers;
@@ -156,6 +180,8 @@ public final class FleetApp {
this.auth = auth;
this.metrics = metrics;
this.memberCredentials = memberCredentials != null ? memberCredentials : MemberCredentialPolicyView::absent;
this.quarantine = quarantine != null ? quarantine : FleetMcp.QuarantineSource.none();
this.outage = outage != null ? outage : FleetMcp.OutageSource.none();
}
/** Wire routes onto a fresh, unstarted Javalin instance. Caller starts it. */
@@ -338,8 +364,15 @@ public final class FleetApp {
if (!allow(ctx, routeAction("GET /agents"), null)) {
return;
}
ctx.status(200).json(Map.of("agents",
workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
try {
ctx.status(200).json(Map.of("agents",
workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
} catch (HerdrException e) {
// fleetd #297: workers.list() reaches herdr — a transport failure must land in the same
// {error, detail} envelope every other failure path here uses, not escape as a bare
// exception and leave Javalin's default handling to respond outside the JSON contract.
herdrError(ctx, e);
}
}
/** CB-304: bridge-owned roster merged with live herdr status by paneId. */
@@ -347,40 +380,57 @@ public final class FleetApp {
if (!allow(ctx, routeAction("GET /members"), null)) {
return;
}
// CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
Map<String, Agent> live = workers.list().stream()
.map(Agent.class::cast)
.filter(a -> a.terminalId() != null)
.collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
// fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
// uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
List<Map<String, Object>> out = sessions.rosterResolved().stream()
.map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
.toList();
Map<String, Object> body = new LinkedHashMap<>();
// fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
// "workers", so a caller that read "members" saw an empty fleet and reported no members at
// all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
// REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
// drops reads this endpoint. Drop the alias once nothing reads it.
body.put("members", out);
body.put("workers", out);
// CB-586: operator visibility for the refs/wip snapshot store without shelling into the
// repo — how many snapshot refs exist and roughly what they cost. Present only once a
// worktree session has established the repo, so a never-snapshotted fleet reports nothing.
sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
Map.of("count", st.count(), "costBytes", st.costBytes())));
ctx.status(200).json(body);
try {
// CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
Map<String, Agent> live = workers.list().stream()
.map(Agent.class::cast)
.filter(a -> a.terminalId() != null)
.collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
// fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
// uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
List<Map<String, Object>> out = sessions.rosterResolved().stream()
.map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
.toList();
Map<String, Object> body = new LinkedHashMap<>();
// fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
// "workers", so a caller that read "members" saw an empty fleet and reported no members at
// all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
// REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
// drops reads this endpoint. Drop the alias once nothing reads it.
body.put("members", out);
body.put("workers", out);
// CB-586: operator visibility for the refs/wip snapshot store without shelling into the
// repo — how many snapshot refs exist and roughly what they cost. Present only once a
// worktree session has established the repo, so a never-snapshotted fleet reports nothing.
sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
Map.of("count", st.count(), "costBytes", st.costBytes())));
ctx.status(200).json(body);
} catch (HerdrException e) {
// fleetd #297: same reasoning as agents() above — this is the out-of-band roster a lead
// falls back to when its MCP mount drops, so it must stay inside the JSON error contract
// exactly when herdr is briefly unreachable, not escape as a bare exception.
herdrError(ctx, e);
}
}
/** The configured worker profiles and which one a no-argument spawn uses. */
/**
* The configured worker profiles, which one a no-argument spawn uses, and (fleetd #297) the two
* outage states {@code fleet_profiles} already reports: {@code quarantined} (CB-578 stage B —
* the backend reported it out of capacity) and {@code coolingOff} (fleetd #201 Unit 5 — the
* credential threw repeated non-exhaustion backend errors). Both are read from the SAME shared
* {@link FleetMcp.QuarantineSource}/{@link FleetMcp.OutageSource} instances {@code FleetMcp}
* reads, never recomputed, so the two doors cannot disagree about which profile is down and why.
* Independent checks, so a profile can appear in both maps at once; each map is present only
* when at least one profile is in that state.
*/
private void profiles(Context ctx) {
if (!allow(ctx, routeAction("GET /profiles"), null)) {
return;
}
ctx.status(200).json(Map.of(
"profiles", workers.profiles(),
"default", workers.defaultProfile() == null ? "" : workers.defaultProfile()));
// fleetd #297: ONE body builder, shared with fleet_profiles. Handing both doors the same
// QuarantineSource/OutageSource instances stops them reading different facts; rendering
// through the same method stops them reporting those facts differently. Both are needed.
ctx.status(200).json(FleetMcp.profilesView(workers, quarantine, outage));
}
/**
@@ -634,7 +684,19 @@ public final class FleetApp {
ctx.status(400).json(Map.of("error", "bad_request", "detail", "body must be JSON"));
return;
}
messages.reply(id, content);
// fleetd #302: content is required. `.path("content").asText("")` above turns a missing key
// into "" rather than throwing, so without this check an empty/blank reply used to reach
// messages.reply(...) and silently resolve the lead's waiter — the same class of bug as the
// sibling "content is required" guards on sendMessage/askMessage below, except this one wrote
// a WRONG value instead of failing loudly. The check lives in MessageService.reply so both
// this door and FleetMcp.reply inherit the same rule; this catch only translates it into the
// {error, detail} envelope this file uses everywhere else.
try {
messages.reply(id, content);
} catch (IllegalArgumentException e) {
ctx.status(400).json(Map.of("error", "bad_request", "detail", e.getMessage()));
return;
}
ctx.status(200).json(Map.of("sessionId", id, "delivered", true));
}
@@ -864,10 +864,20 @@ public final class SessionManager implements TurnListener {
}
/**
* Gracefully drain all registered sessions on daemon shutdown. For each session that is
* {@code BUSY}, poll up to {@code timeoutNanos} for it to leave {@code BUSY}, then release it
* regardless. Non-busy sessions are released immediately. A failure releasing one session is
* logged and does not abort the rest.
* Gracefully drain all registered sessions on daemon shutdown. Non-busy sessions are released
* immediately; a {@code BUSY} one is polled until it leaves {@code BUSY}, then released
* regardless. A failure releasing one session is logged and does not abort the rest.
*
* <p>{@code timeoutNanos} is a budget for the WHOLE drain, not a grace period per session: the
* deadline is taken once, before the loop. So the first BUSY session can spend all of it, and a
* later BUSY one is then released with no wait at all. That is deliberate. This drain is only
* one phase of shutdown — {@code Fleetd} closes the message service, the push loop, the
* heartbeat, MCP and the router after it — and the whole sequence has to finish inside
* launchd's exit window. A per-session grace would let N busy members drain for N * the
* timeout, overrun that window, and get the daemon SIGKILLed part-way through; the members not
* yet reached would then get no clean release, no preserved-worktree log, and no snapshot.
* Cutting one turn short is the cheaper failure, and it is not silent: an abandoned BUSY
* session is preserved, snapshotted, and logged at WARN by {@code logPreservedForShutdown}.
*
* <p>CB-544: this is a {@link ReleaseCause#SHUTDOWN} release — the worker's pane is stopped
* (the process must end) but its worktree is preserved and its path logged. Shutdown is never
@@ -40,6 +40,7 @@ public final class FakeHerdr implements HerdrClient {
private final Map<String, List<String>> extraTabs = new LinkedHashMap<>();
private int agentNameTakenFor = 0;
private int agentPaneBusyFor = 0;
private String agentStartErrorCode = null;
private int workerTabPaneCount = 1;
private String paneCloseErrorCode = null;
private final Map<String, String> paneCloseErrorCodeFor = new ConcurrentHashMap<>();
@@ -84,6 +85,12 @@ public final class FakeHerdr implements HerdrClient {
return this;
}
/** Make every {@code agent.start} call fail with this herdr error code. */
public FakeHerdr agentStartFailsWith(String code) {
this.agentStartErrorCode = code;
return this;
}
/** Make the worker tab (w9:t2) report this many panes in {@code tab.list} (default 1). */
public FakeHerdr withWorkerTabPaneCount(int n) {
this.workerTabPaneCount = n;
@@ -311,6 +318,10 @@ public final class FakeHerdr implements HerdrClient {
+ required + "`", "invalid_request", null);
}
}
if (agentStartErrorCode != null) {
throw new HerdrException("herdr error [" + agentStartErrorCode + "]: agent.start failed",
agentStartErrorCode, null);
}
long starts = calls.stream().filter(c -> c.method().equals("agent.start")).count();
if (starts <= agentPaneBusyFor) {
throw new HerdrException(
@@ -1165,7 +1165,8 @@ class ClaudeCodeLauncherTest {
@Test
void spawnLetsAnUnrelatedHerdrErrorPropagateUnchanged() {
// Fix 1 must only special-case a "*_not_found" answer. Any other herdr failure keeps
// propagating as-is — this gate does not know how to recover from it.
// propagating as-is — this gate does not know how to recover from it. The pane still needs
// closing because spawn throws before it can return the pane id to a caller that could stop it.
FakeHerdr herdr = new FakeHerdr();
herdr.agentStatus("unknown");
herdr.agentGetFailsWithAfter(0, "internal_error");
@@ -1182,8 +1183,27 @@ class ClaudeCodeLauncherTest {
() -> svc.spawn(new SpawnRequest(null, null, null)));
assertEquals("internal_error", ex.code());
assertEquals(0, paneCloseCount(herdr, "w9:pRoot_1"),
"an error this gate does not recognize is not this gate's teardown to run");
assertEquals(1, paneCloseCount(herdr, "w9:pRoot_1"),
"the unchanged error leaves spawn without a pane id, so this gate closes its orphaned pane");
}
@Test
void panePlacementClosesTheSplitPaneWhenAgentStartFails() {
FakeHerdr herdr = new FakeHerdr().agentStartFailsWith("internal_error");
FleetConfig.Profile cfg = new FleetConfig.Profile(
"ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
List.of("claude"), "pane", "fleetd-workers", "w #{n}", null, 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);
dev.ltms.fleet.herdr.HerdrException ex = assertThrows(
dev.ltms.fleet.herdr.HerdrException.class,
() -> svc.spawn(new SpawnRequest(null, null, null)));
assertEquals("internal_error", ex.code(), "agent.start failure propagates unchanged");
assertEquals(1, paneCloseCount(herdr, "w1:pSplit"),
"the pane split for a peer that never starts is closed instead of left orphaned");
}
// --- fleetd #176 fix 2: corroborated UNKNOWN refinement --------------------------------------
@@ -19,6 +19,10 @@ import dev.ltms.fleet.session.SessionManager;
import dev.ltms.fleet.session.Worktrees;
import dev.ltms.fleet.member.ClaudeCodeLauncher;
import dev.ltms.fleet.member.CompositePeerLauncher;
import dev.ltms.fleet.member.MemberCredentialPolicyView;
import dev.ltms.fleet.mcp.FleetMcp;
import dev.ltms.fleet.placement.BackendOutagePolicy;
import dev.ltms.fleet.placement.BackendQuarantine;
import dev.ltms.fleet.placement.PlacementPolicies;
import io.javalin.Javalin;
import org.junit.jupiter.api.AfterEach;
@@ -32,6 +36,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import static org.junit.jupiter.api.Assertions.*;
@@ -70,6 +75,19 @@ class FleetAppTest {
private int start(FakeHerdr herdr, String workerBaseUrl, Set<String> allow, String placement,
Worktrees worktrees, Predicate<String> deliverable) {
return start(herdr, workerBaseUrl, allow, placement, worktrees, deliverable,
FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
}
/**
* fleetd #297: same wiring as above, plus the two SAME shared sources {@code GET /profiles}
* must read — lets a test prove the quarantined/coolingOff facts it reports come from a real
* {@link dev.ltms.fleet.placement.BackendQuarantine}/{@link
* dev.ltms.fleet.placement.BackendOutagePolicy}, exactly like {@code fleet_profiles}'s own tests.
*/
private int start(FakeHerdr herdr, String workerBaseUrl, Set<String> allow, String placement,
Worktrees worktrees, Predicate<String> deliverable,
FleetMcp.QuarantineSource quarantine, FleetMcp.OutageSource outage) {
FleetConfig.Profile wcfg = new FleetConfig.Profile(
"ltms-local", workerBaseUrl, "coder", null, "FLEETD_WORKER_TOKEN", null,
placement, "fleet", "worker: {profile} #{n}", null, null, null);
@@ -90,8 +108,9 @@ class FleetAppTest {
// it directly so the inbox contract holds for those endpoints.
inbox.own("term_a");
MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
app = new FleetApp(herdr, workers, sessions, messages, this.presence, null,
null, null, id -> this.presence.isPresent(id) || deliverable.test(id))
app = new FleetApp(herdr, herdr, workers, sessions, messages, this.presence, null,
null, null, id -> this.presence.isPresent(id) || deliverable.test(id),
MemberCredentialPolicyView::absent, quarantine, outage)
.build().start("127.0.0.1", 0);
return app.port();
}
@@ -164,6 +183,22 @@ class FleetAppTest {
assertEquals("idle", agents.get(0).get("status").asText());
}
/**
* fleetd #297 gap 1: {@code workers.list()} reaches herdr, and a transport failure there must
* land in the same {@code {error, detail}} envelope every other failure path in this file uses
* (see {@code herdrError}), not escape as a bare exception outside the JSON contract.
*/
@Test
void agentsMapsAHerdrFailureToTheJsonErrorEnvelope() throws Exception {
FakeHerdr down = new FakeHerdr().healthy(false);
int port = start(down, "http://gx00.gw:8000", Set.of("gx00.gw"));
HttpResponse<String> res = req(port, "GET", "/agents");
assertEquals(502, res.statusCode(), res.body());
JsonNode body = mapper.readTree(res.body());
assertEquals("herdr_error", body.get("error").asText());
assertTrue(body.has("detail"), res.body());
}
@Test
void spawnWorkerLandsInOwnTabInWorkerSpaceAndInjectsBaseUrl() throws Exception {
FakeHerdr herdr = new FakeHerdr();
@@ -203,6 +238,42 @@ class FleetAppTest {
JsonNode body = mapper.readTree(req(port, "GET", "/profiles").body());
assertEquals("ltms-local", body.get("default").asText());
assertEquals("ltms-local", body.get("profiles").get(0).asText());
assertFalse(body.has("quarantined"), "nothing is quarantined, so the key is omitted: " + body);
assertFalse(body.has("coolingOff"), "nothing is cooling off, so the key is omitted: " + body);
}
/**
* fleetd #297 gap 2: {@code GET /profiles} must report the same two outage states {@code
* fleet_profiles} does — CB-578 stage B exhaustion quarantine and fleetd #201 Unit 5 cool-off —
* reading the SAME shared {@link BackendQuarantine}/{@link BackendOutagePolicy} instances rather
* than recomputing them. The two checks are independent, and this profile is deliberately put in
* both states at once, matching {@code FleetMcpTest}'s own coverage of that overlap.
*/
@Test
void profilesReportsQuarantineAndCoolingOffFromTheSameSharedSources() throws Exception {
FakeHerdr herdr = new FakeHerdr();
BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
quarantine.quarantine("shared-openai");
FleetMcp.QuarantineSource quarantineSource = new FleetMcp.QuarantineSource(
profile -> "ltms-local".equals(profile) ? "shared-openai" : null, quarantine);
BackendOutagePolicy outagePolicy = new BackendOutagePolicy(() -> 0L);
outagePolicy.record("shared-openai", "t1", "API Error: rate limited");
outagePolicy.record("shared-openai", "t2", "API Error: rate limited"); // 2nd distinct target starts the incident
FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(
profile -> "ltms-local".equals(profile) ? "shared-openai" : null, outagePolicy);
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"), "tab", new GitWorktrees(),
ignored -> false, quarantineSource, outageSource);
JsonNode body = mapper.readTree(req(port, "GET", "/profiles").body());
assertTrue(body.has("quarantined"), body.toString());
assertEquals("shared-openai",
body.get("quarantined").get("ltms-local").get("credentialId").asText());
assertEquals(1800,
body.get("quarantined").get("ltms-local").get("quarantinedForSeconds").asLong());
assertTrue(body.has("coolingOff"), body.toString());
assertEquals("shared-openai",
body.get("coolingOff").get("ltms-local").get("credentialId").asText());
assertEquals(60, body.get("coolingOff").get("ltms-local").get("coolingOffForSeconds").asLong());
}
@Test
@@ -239,6 +310,23 @@ class FleetAppTest {
"liveStatus is unknown when herdr has no matching pane");
}
/**
* fleetd #297 gap 1: same reasoning as {@code agentsMapsAHerdrFailureToTheJsonErrorEnvelope} —
* {@code GET /members} is the endpoint's own comment names as "the out-of-band path a lead falls
* back to when its MCP mount drops", so it must stay inside the {@code {error, detail}} envelope
* exactly when herdr is briefly unreachable.
*/
@Test
void membersMapsAHerdrFailureToTheJsonErrorEnvelope() throws Exception {
FakeHerdr down = new FakeHerdr().healthy(false);
int port = start(down, "http://gx00.gw:8000", Set.of("gx00.gw"));
HttpResponse<String> res = req(port, "GET", "/members");
assertEquals(502, res.statusCode(), res.body());
JsonNode body = mapper.readTree(res.body());
assertEquals("herdr_error", body.get("error").asText());
assertTrue(body.has("detail"), res.body());
}
@Test
void spawnWithACwdParamRootsTheWorkerThere() throws Exception {
FakeHerdr herdr = new FakeHerdr();
@@ -444,6 +532,53 @@ class FleetAppTest {
assertEquals("orphan", body.get("replies").get(0).get("content").asText());
}
@Test
void replyWithMissingContentIsRejectedAndDoesNotResolveTheWaiter() throws Exception {
// fleetd #302: `.path("content").asText("")` used to turn a missing "content" key into an
// empty string that reached rendezvous.resolve, silently completing the lead's blocking wait
// with nothing. Prove the fix two ways: the bad call is rejected with 400, AND the real send
// it would have wrongly resolved is still open afterwards — a real reply completes it.
FakeHerdr herdr = new FakeHerdr().agentStatus("idle");
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
var send = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
try { return postMessage(port, "{\"content\":\"review this\",\"timeoutMs\":4000}"); }
catch (Exception e) { throw new RuntimeException(e); }
});
Thread.sleep(200); // let the background send open its rendezvous waiter
HttpResponse<String> badReply = postJson(port, "/sessions/term_a/reply", "{}");
assertEquals(400, badReply.statusCode());
JsonNode err = mapper.readTree(badReply.body());
assertEquals("bad_request", err.get("error").asText());
assertTrue(err.has("detail"));
// The waiter must still be open — a real reply now completes the ORIGINAL send.
HttpResponse<String> goodReply = postJson(port, "/sessions/term_a/reply", "{\"content\":\"LGTM ship it\"}");
assertEquals(200, goodReply.statusCode());
HttpResponse<String> res = send.get(6, java.util.concurrent.TimeUnit.SECONDS);
assertEquals(200, res.statusCode());
assertEquals("LGTM ship it", mapper.readTree(res.body()).get("reply").asText());
}
@Test
void replyWithEmptyOrWhitespaceContentIsRejectedSameAsMissing() throws Exception {
// fleetd #302 sibling case: present-but-blank content is treated the same as a missing key —
// FleetMcp's own required-content guard (fleet_reply's "content is required") makes no
// distinction between the two either, so diverging here would be a new asymmetry.
int port = startHealthy();
HttpResponse<String> empty = postJson(port, "/sessions/term_a/reply", "{\"content\":\"\"}");
assertEquals(400, empty.statusCode());
assertEquals("bad_request", mapper.readTree(empty.body()).get("error").asText());
HttpResponse<String> whitespace = postJson(port, "/sessions/term_a/reply", "{\"content\":\" \"}");
assertEquals(400, whitespace.statusCode());
assertEquals("bad_request", mapper.readTree(whitespace.body()).get("error").asText());
}
@Test
void drainRepliesReturnsEmptyForNoReplies() throws Exception {
int port = startHealthy();