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 be5ba22c75 #298: AmqpReplyInbox.release() requeues held deliveries instead of dropping them
CI / contract (pull_request) Successful in 1m12s
CI / build (pull_request) Successful in 2m3s
release(target) used to cancel the target's consumer and clear held's local
record for it. Cancelling a consumer does not requeue the broker's in-flight
deliveries — they stay unacked on the still-open channel until a real
connection drop. So a held-but-undrained reply became permanently
unreachable: never acked, never nacked, never requeued, invisible to peek.

Fix: cancel the consumer first (so it can no longer receive redeliveries),
then nack-with-requeue every held delivery for that target before dropping
the local record. Nacking before the cancel was tried first but a real
broker demonstrated a race: the still-active consumer immediately received
the requeued message back, racing held.remove and leaving peek non-empty.
Cancelling first avoids that. A failed requeue is logged at WARN and does
not abort release(), matching the best-effort teardown style #293 settled
for HerdrPeerLauncher.stop().

Extends AmqpReplyInboxContractTest.releaseCancelsConsumer... to assert the
held delivery is recoverable via a later own(), not just absent from peek.
2026-09-04 12:14: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
8 changed files with 379 additions and 55 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;
}
/**
@@ -208,18 +208,72 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
}
}
/**
* Release ownership of {@code target}: cancel its consumer, then nack-with-requeue every
* delivery still held for it instead of just dropping the local record.
*
* <p><strong>Cancelling a consumer does not requeue its in-flight deliveries.</strong> In AMQP,
* a delivery that was pushed to a consumer stays unacked, attached to the still-open
* {@link #channel}, until that channel or the connection closes — {@code basicCancel} alone does
* neither. So before this method existed with a requeue step, it dropped {@link #held}'s entries
* for {@code target} while the broker still considered them outstanding: never acked, never
* nacked, never requeued, and no longer reachable by {@link #peek} — permanently invisible. This
* is unlike {@link #handleRecovery} and {@link #close()}, whose bare {@code held.clear()} is
* correct because each has already made the broker requeue (a real connection drop, or
* {@code channel.close()} respectively) before clearing local state.
*
* <p><strong>Order: cancel first, then nack.</strong> A delivery tag stays valid for
* {@code basicNack} on this channel regardless of whether its consumer is still attached — only
* a channel/connection close invalidates it — so cancelling {@code target}'s consumer first does
* not risk the tags. Doing it the other way round does: nacking a delivery with {@code requeue}
* while its consumer is still active hands the message straight back to that <em>same</em>
* consumer the instant a prefetch slot frees up (confirmed against a real broker — see
* {@code AmqpReplyInboxContractTest.releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery}),
* which races this method's own {@code held.remove(target)}: the redelivery can land after the
* clear and leave a stale entry behind, so {@link #peek} is no longer reliably empty right after
* {@link #release}. Cancelling first closes that consumer, so the requeued message goes back to
* the queue for whichever consumer picks it up next (a later {@link #own}), not this one.
*
* <p><strong>Failure of the requeue is best-effort, not fatal.</strong> {@link #release} runs
* during teardown ({@code Fleetd} calls it right after {@code MessageService.abandon}), and a
* throw here would abort cleanups the caller depends on — the same argument fleetd #293 settled
* for {@code HerdrPeerLauncher.stop()}'s tab-close step. So a failed {@code basicNack} is logged
* at WARN, naming the target and delivery tag that leaked, and release proceeds; the delivery
* stays unacked on the broker rather than being silently dropped, so it is still recoverable by a
* later connection drop even though this release did not manage to requeue it immediately. A
* failed {@code basicCancel} still throws, unchanged from before this fix — that failure means
* the consumer may still be attached, so best-effort requeue is not attempted underneath it.
*/
@Override
public void release(String target) {
synchronized (channelLock) {
String tag = consumerTags.remove(target);
held.remove(target); // stale delivery tags must not survive release
if (tag == null) {
return;
if (tag != null) {
try {
channel.basicCancel(tag);
} catch (IOException e) {
throw new IllegalStateException("cannot cancel consumer for " + target, e);
}
}
try {
channel.basicCancel(tag);
} catch (IOException e) {
throw new IllegalStateException("cannot cancel consumer for " + target, e);
var perTarget = held.remove(target);
if (perTarget != null) {
synchronized (perTarget) {
for (Held h : perTarget.values()) {
try {
channel.basicNack(h.deliveryTag(), false, true); // requeue, don't drop
} catch (IOException | RuntimeException e) {
// Caught broadly (not just IOException) for the same reason #293 catches
// RuntimeException in HerdrPeerLauncher.stop(): best-effort teardown must
// not be guarded only against the expected failure and bare against any
// other. The message stays unacked on the broker either way — not lost,
// just not proactively requeued — until a connection drop frees it.
log.warn("release({}): could not requeue held delivery (msgId={}, tag={})"
+ " back to the broker — it stays unacked until a connection"
+ " drop frees it: {}",
target, h.message().msgId(), h.deliveryTag(), e.getMessage());
}
}
}
}
}
}
@@ -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
@@ -154,7 +154,7 @@ class AmqpReplyInboxContractTest {
}
@Test
void releaseCancelsConsumerAndClearsHeld() throws Exception {
void releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery() throws Exception {
String target = "worker-release-" + System.nanoTime();
try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
inbox.own(target);
@@ -164,6 +164,22 @@ class AmqpReplyInboxContractTest {
inbox.release(target);
assertTrue(inbox.peek(target).isEmpty(),
"release clears the local held snapshot");
// fleetd #298: release() must not just drop the local record — the broker delivery was
// never acked, so cancelling the consumer alone leaves it unacked-but-orphaned on the
// still-open channel unless release() nacks it back with requeue=true. Prove the message
// is genuinely recoverable, not merely absent from peek: re-own the same target and
// confirm the broker redelivers it to the fresh consumer.
inbox.own(target);
List<ReplyInbox.InboxMessage> recovered = awaitPeek(inbox, target);
assertEquals(1, recovered.size(),
"a reply held (but undrained) at release() time must still be recoverable — "
+ "release() must requeue it, not silently drop it while the broker still "
+ "considers it outstanding");
assertEquals("m1", recovered.getFirst().msgId());
assertEquals("release me", recovered.getFirst().content());
inbox.ack(target, "m1");
}
}
@@ -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();