Compare commits
32 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bbbb4c1eb3 | |||
| 38c248e617 | |||
| 9debc0de27 | |||
| f34361b263 | |||
| be5ba22c75 | |||
| 85c90d440a | |||
| e97502d550 | |||
| aef14ff46e | |||
| cba516bda4 | |||
| ba51e0c6cc | |||
| 086c59848e | |||
| 0c10079755 | |||
| ece2091b53 | |||
| b3f917e6f5 | |||
| fa39a5f55e | |||
| d5128a1d35 | |||
| 61097e5cf0 | |||
| ef507bcd12 | |||
| 94f50e507a | |||
| 75b15086b0 | |||
| dab9645906 | |||
| e93b5f6512 | |||
| bfabe13e8f | |||
| 7e49c6eca2 | |||
| 11cbfa79b4 | |||
| c2c2746922 | |||
| 6ed70700a0 | |||
| 0087645da4 | |||
| 19cacf5b62 | |||
| 4dd12083ab | |||
| 30e21adec7 | |||
| 719b79f892 |
@@ -250,9 +250,7 @@ public final class Fleetd {
|
||||
boolean clearAfterTurn = cfg.lifecycle() != null && cfg.lifecycle().clearAfterTurn();
|
||||
SessionManager sessions = new SessionManager(workers, new GitWorktrees(cfg.worktreeRoot(), cfg.worktreeGroup()),
|
||||
System::nanoTime, contextCap, clearAfterTurn);
|
||||
liveCountRef.set(profileName -> (int) sessions.roster().stream()
|
||||
.filter(s -> profileName.equals(s.profile()))
|
||||
.count());
|
||||
liveCountRef.set(profileName -> liveSessionCount(sessions.roster(), profileName));
|
||||
|
||||
// CB-303 part 1: idle-ttl reaper — only when configured, defaults to disabled.
|
||||
final SessionReaper reaper;
|
||||
@@ -627,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 -> {
|
||||
@@ -638,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
|
||||
@@ -717,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);
|
||||
@@ -859,6 +867,19 @@ public final class Fleetd {
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Count sessions that occupy a profile's spawn capacity. A {@code BACKEND_ERROR} or
|
||||
* {@code FAILED} session stays in the roster so {@code fleet_list} can show its failure, but a
|
||||
* member that cannot accept another delivery does not use a seat.
|
||||
*/
|
||||
static int liveSessionCount(List<MemberSession> roster, String profileName) {
|
||||
return (int) roster.stream()
|
||||
.filter(session -> profileName.equals(session.profile()))
|
||||
.filter(session -> session.state() != MemberSession.State.BACKEND_ERROR)
|
||||
.filter(session -> session.state() != MemberSession.State.FAILED)
|
||||
.count();
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #248 / fleetd#201 Unit 5: factory for the production {@link BackendErrorSink} — the
|
||||
* collaborator {@link CompletionResolver} notifies when a pane-scrape classification actually
|
||||
|
||||
@@ -322,8 +322,9 @@ public record FleetConfig(
|
||||
* statement that does not stop being true just because the profile was
|
||||
* named directly. A negative value has no sane meaning (there is no
|
||||
* "excluded" to degrade to below zero) and is refused at config load
|
||||
* instead, naming the profile and the key. Live means any session the
|
||||
* registry still owns (acquired and not yet released), in any state.
|
||||
* instead, naming the profile and the key. Live means a session that can
|
||||
* receive another delivery. The roster keeps terminal {@code BACKEND_ERROR}
|
||||
* and {@code FAILED} sessions for diagnostics, but they do not use capacity.
|
||||
* @param kind which peer launcher spawns this profile: {@code "claude-code"} (default —
|
||||
* the {@link dev.ltms.fleet.member.ClaudeCodeLauncher}) or {@code "opencode"}.
|
||||
* The {@code CompositePeerLauncher} routes {@code spawn}/reap by this value, so
|
||||
|
||||
@@ -14,6 +14,7 @@ import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.BiConsumer;
|
||||
@@ -28,6 +29,14 @@ public final class FleetHealthMonitor {
|
||||
static final int MAX_FAIL_TARGET_ATTEMPTS = 3;
|
||||
// CB-641: Match the injector's 60s readiness gate so health allows a full first boot.
|
||||
static final long READINESS_GRACE_NANOS = TimeUnit.SECONDS.toNanos(60);
|
||||
/**
|
||||
* fleetd #280: how long after a terminal transition to wait before the one bounded re-check
|
||||
* fires. Must exceed the worst-case reverse-rendezvous {@code fleet_ask} window (55-115s, see
|
||||
* {@code FleetMcp.ASK_DEFAULT_TIMEOUT_MS} / {@code FleetApp.MAX_ASK_TIMEOUT_MS}) so that, if the
|
||||
* target was genuinely {@code ASKING} when {@code state} was first observed, its own ask has had
|
||||
* time to lapse (clearing {@code Task#question} back to {@code null}) before this fires.
|
||||
*/
|
||||
static final long ASK_LAPSE_RECHECK_DELAY_SECONDS = 120;
|
||||
|
||||
private final AgentControl agents;
|
||||
private final Supplier<List<MemberSession>> roster;
|
||||
@@ -38,7 +47,16 @@ public final class FleetHealthMonitor {
|
||||
private final long workingSuspectAfterNanos;
|
||||
private final BiConsumer<String, String> failTarget;
|
||||
private final Map<String, HealthPrior> priors = new HashMap<>();
|
||||
private final Map<String, HealthState> states = new HashMap<>();
|
||||
/**
|
||||
* The live classification per member, and the only one of this class's three maps that more
|
||||
* than one scheduler task touches. {@code tick} writes it (and prunes it to the roster);
|
||||
* fleetd #280's delayed {@link #recheckTerminalTarget} reads it from its own separate scheduled
|
||||
* task. Both run on the single-threaded scheduler {@code Fleetd} passes in today, so they are
|
||||
* serialised — but nothing in this class enforces that, and an unsynchronised {@link HashMap}
|
||||
* read racing a resize can spin a CPU forever rather than fail visibly. {@code priors} and
|
||||
* {@code orphanStreaks} stay plain maps because {@code tick} is still their only toucher.
|
||||
*/
|
||||
private final Map<String, HealthState> states = new ConcurrentHashMap<>();
|
||||
/**
|
||||
* CB-643: consecutive ticks on which a target looked like an orphaned delegation. The fact
|
||||
* {@link MessageService#hasOrphanedDelegation} reports is a true snapshot, but it can read true
|
||||
@@ -171,6 +189,7 @@ public final class FleetHealthMonitor {
|
||||
// member stayed terminal.
|
||||
if (terminal(next)) {
|
||||
failTerminalTarget(target, next);
|
||||
scheduleTerminalRecheck(target, next);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -191,6 +210,48 @@ public final class FleetHealthMonitor {
|
||||
target, state, MAX_FAIL_TARGET_ATTEMPTS, last);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #280: schedule the one bounded, delayed follow-up for a terminal transition — never a
|
||||
* per-tick retry (CB-580 rejected that shape; {@link #reportTransition} still fires
|
||||
* {@link #failTerminalTarget} exactly once per transition, unconditionally on the tick loop).
|
||||
* This is a single one-shot task, scheduled once per transition into GONE/NEVER_READY, so a
|
||||
* member stuck terminal for the rest of its life gets exactly one extra attempt, not one per
|
||||
* tick. See {@link #recheckTerminalTarget} for why the extra attempt is safe.
|
||||
*/
|
||||
private void scheduleTerminalRecheck(String target, HealthState state) {
|
||||
if (scheduler.isShutdown()) return;
|
||||
try {
|
||||
scheduler.schedule(() -> recheckTerminalTarget(target, state),
|
||||
ASK_LAPSE_RECHECK_DELAY_SECONDS, TimeUnit.SECONDS);
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("fleet health: could not schedule terminal re-check for member={} state={}",
|
||||
target, state, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #280: the delayed re-check {@link #scheduleTerminalRecheck} scheduled for one terminal
|
||||
* transition. By now, a {@code fleet_ask} that was still open when {@code state} was first
|
||||
* observed has had time to lapse on its own (see {@link #ASK_LAPSE_RECHECK_DELAY_SECONDS}),
|
||||
* clearing {@code Task#question} back to {@code null} — which is exactly what
|
||||
* {@link MessageService#abandon(String, String, boolean)}'s {@code sweepAsking=false} filter
|
||||
* needs to finally match it. Calling {@link #failTerminalTarget} again is safe only because
|
||||
* {@code sweepAsking} stays {@code false}: a task genuinely still {@code ASKING} is skipped
|
||||
* exactly as it was on the very first attempt — this never fails a ticket whose ask has not yet
|
||||
* lapsed.
|
||||
*
|
||||
* <p><strong>Guarded on "target is still classified {@code state}."</strong> Without this guard,
|
||||
* a member that recovered (or was released and dropped from the roster) between the transition
|
||||
* and this re-check would still take a blind {@code failTarget} call — reaching into whatever
|
||||
* brand-new, unrelated turn it has since picked up and failing it too. {@link #states} already
|
||||
* carries the live classification (updated every tick, pruned to the current roster on release),
|
||||
* so a stale or recovered target simply reads as a mismatch here and this is a no-op.
|
||||
*/
|
||||
void recheckTerminalTarget(String target, HealthState state) {
|
||||
if (states.get(target) != state) return;
|
||||
failTerminalTarget(target, state);
|
||||
}
|
||||
|
||||
private static boolean terminal(HealthState state) {
|
||||
return state == HealthState.GONE || state == HealthState.NEVER_READY;
|
||||
}
|
||||
|
||||
@@ -257,7 +257,7 @@ public final class FleetMcp {
|
||||
// Each handler is built once and wired to its fleet_* tool below.
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> sendHandler =
|
||||
(exchange, req) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SEND,
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_send", req.arguments()),
|
||||
str(req.arguments(), "sessionId"));
|
||||
if (denied != null) return denied;
|
||||
String caller = callerTerminal(exchange);
|
||||
@@ -299,7 +299,7 @@ public final class FleetMcp {
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> replyHandler =
|
||||
(exchange, req) -> {
|
||||
String self = callerTerminal(exchange);
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.REPLY, self);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_reply", req.arguments()), self);
|
||||
if (denied != null) return denied;
|
||||
return reply(messages, self, str(req.arguments(), "content"));
|
||||
};
|
||||
@@ -307,13 +307,13 @@ public final class FleetMcp {
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> askHandler =
|
||||
(exchange, req) -> {
|
||||
String self = callerTerminal(exchange);
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.ASK, self);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_ask", req.arguments()), self);
|
||||
if (denied != null) return denied;
|
||||
return ask(messages, self, str(req.arguments(), "question"), timeoutMs(req.arguments()));
|
||||
};
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> statusHandler =
|
||||
(exchange, req) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_status", req.arguments()), null);
|
||||
if (denied != null) return denied;
|
||||
return status(messages, str(req.arguments(), "sessionId"));
|
||||
};
|
||||
@@ -322,7 +322,7 @@ public final class FleetMcp {
|
||||
Map<String, Object> a = req.arguments();
|
||||
String target = str(a, "target");
|
||||
// The action depends on the ARGUMENTS, not on the tool name -- see pollAction.
|
||||
McpSchema.CallToolResult denied = deny(exchange, pollAction(target), target);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_poll", a), target);
|
||||
if (denied != null) return denied;
|
||||
return poll(messages, str(a, "ticket"), target);
|
||||
};
|
||||
@@ -331,14 +331,14 @@ public final class FleetMcp {
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> ackHandler =
|
||||
(exchange, req) -> {
|
||||
Map<String, Object> a = req.arguments();
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.DRAIN, str(a, "target"));
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_ack", a), str(a, "target"));
|
||||
if (denied != null) return denied;
|
||||
return ack(messages, str(a, "target"), str(a, "msgId"));
|
||||
};
|
||||
// Fleet management (CB-108): spawn/list/stop over ClaudeCodeLauncher.
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> spawnHandler =
|
||||
(exchange, req) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SPAWN, null);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_spawn", req.arguments()), null);
|
||||
if (denied != null) return denied;
|
||||
String caller = callerTerminal(exchange);
|
||||
// SPAWN is already auth-gated to PRIMARY (architects can never call it), but
|
||||
@@ -355,7 +355,7 @@ public final class FleetMcp {
|
||||
};
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> listHandler =
|
||||
(exchange, _) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_list", Map.of()), null);
|
||||
if (denied != null) return denied;
|
||||
return listFleet(workers, sessions, messages, capacity, healthCoverage, quarantine, outage,
|
||||
leadSeats, callers == null ? Map.of() : callers.leads(),
|
||||
@@ -365,19 +365,19 @@ public final class FleetMcp {
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> stopHandler =
|
||||
(exchange, req) -> {
|
||||
String paneId = str(req.arguments(), "paneId");
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.STOP, paneId);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_stop", req.arguments()), paneId);
|
||||
if (denied != null) return denied;
|
||||
return stop(sessions, paneId);
|
||||
};
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> profilesHandler =
|
||||
(exchange, _) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_profiles", Map.of()), null);
|
||||
if (denied != null) return denied;
|
||||
return profiles(workers, quarantine, outage);
|
||||
};
|
||||
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> whoamiHandler =
|
||||
(exchange, _) -> {
|
||||
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
|
||||
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_whoami", Map.of()), null);
|
||||
if (denied != null) return denied;
|
||||
return whoami(principal(exchange), sessions);
|
||||
};
|
||||
@@ -754,6 +754,25 @@ public final class FleetMcp {
|
||||
return isBlank(target) ? Authz.Action.READ : Authz.Action.DRAIN;
|
||||
}
|
||||
|
||||
/**
|
||||
* The action a registered tool handler actually hands to the authorization gate.
|
||||
* Keeping this choice beside the registered-tool inventory makes a new tool fail the coverage
|
||||
* test until its action is pinned.
|
||||
*/
|
||||
static Authz.Action toolAction(String toolName, Map<String, Object> arguments) {
|
||||
return switch (toolName) {
|
||||
case "fleet_send" -> Authz.Action.SEND;
|
||||
case "fleet_reply" -> Authz.Action.REPLY;
|
||||
case "fleet_ask" -> Authz.Action.ASK;
|
||||
case "fleet_status", "fleet_list", "fleet_profiles", "fleet_whoami" -> Authz.Action.READ;
|
||||
case "fleet_poll" -> pollAction(str(arguments, "target"));
|
||||
case "fleet_ack" -> Authz.Action.DRAIN;
|
||||
case "fleet_spawn" -> Authz.Action.SPAWN;
|
||||
case "fleet_stop" -> Authz.Action.STOP;
|
||||
default -> throw new IllegalArgumentException("unregistered tool: " + toolName);
|
||||
};
|
||||
}
|
||||
|
||||
/** {@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)) {
|
||||
@@ -996,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());
|
||||
@@ -1027,7 +1065,7 @@ public final class FleetMcp {
|
||||
if (!coolingOff.isEmpty()) {
|
||||
result.put("coolingOff", coolingOff);
|
||||
}
|
||||
return text(json(result));
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1148,15 +1186,35 @@ public final class FleetMcp {
|
||||
private static Map<String, Object> memberCapacityView(MemberSession session, Agent live,
|
||||
MessageService messages, long nowNanos) {
|
||||
Map<String, Object> row = SessionManager.rosterView(session, live);
|
||||
boolean open = messages != null && messages.hasAcceptedDelivery(session.terminalId());
|
||||
boolean inbox = messages != null && messages.hasInboxMessage(session.terminalId());
|
||||
boolean reclaimable = (session.state() == MemberSession.State.READY || session.state() == MemberSession.State.DONE)
|
||||
&& !open && !inbox;
|
||||
boolean reclaimable = reclaimable(session, messages);
|
||||
row.put("reclaimable", reclaimable);
|
||||
row.put("idleForSeconds", reclaimable ? Math.max(0, (nowNanos - session.lastActivityAtNanos()) / 1_000_000_000L) : null);
|
||||
return row;
|
||||
}
|
||||
|
||||
/**
|
||||
* The one definition of {@code reclaimable}: this member holds a spawn seat, and has no open
|
||||
* bridge work, so stopping it gives the seat back. Both views in a single {@code fleet_list}
|
||||
* response call it — the per-member flag in {@link #memberCapacityView} and the per-profile
|
||||
* count in {@link #capacityView} — because two copies of this rule in one response is how the
|
||||
* two numbers come to disagree.
|
||||
*
|
||||
* <p>fleetd #284: {@code BACKEND_ERROR} and {@code FAILED} are deliberately NOT reclaimable.
|
||||
* The ticket asked for them to be, and that half of the ticket was wrong. Once
|
||||
* {@code Fleetd.liveSessionCount} stopped counting a terminal session as live, that seat is
|
||||
* ALREADY in {@code free}; counting it here too reports the same seat twice, and
|
||||
* {@code free + reclaimable} then reads as more capacity than {@code maxLoad} allows. The dead
|
||||
* session stays visible either way: its roster row still carries {@code state:
|
||||
* "backend_error"} or {@code "failed"}, which is what tells the lead to stop it.
|
||||
*/
|
||||
static boolean reclaimable(MemberSession session, MessageService messages) {
|
||||
boolean holdsSeat = session.state() == MemberSession.State.READY
|
||||
|| session.state() == MemberSession.State.DONE;
|
||||
return holdsSeat && (messages == null
|
||||
|| (!messages.hasAcceptedDelivery(session.terminalId())
|
||||
&& !messages.hasInboxMessage(session.terminalId())));
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-583: {@code free} alone cannot tell a lead "busy, will free up" from "refusing, and
|
||||
* nothing changes for N seconds" — those need different decisions. So a quarantined profile
|
||||
@@ -1195,8 +1253,7 @@ public final class FleetMcp {
|
||||
int live = liveCount.apply(profile);
|
||||
int leadSeatCount = leadSeats.seatsFor().apply(profile);
|
||||
int reclaimable = (int) roster.stream().filter(s -> profile.equals(s.profile()))
|
||||
.filter(s -> (s.state() == MemberSession.State.READY || s.state() == MemberSession.State.DONE))
|
||||
.filter(s -> messages == null || (!messages.hasAcceptedDelivery(s.terminalId()) && !messages.hasInboxMessage(s.terminalId())))
|
||||
.filter(s -> reclaimable(s, messages))
|
||||
.count();
|
||||
Map<String, Object> row = new LinkedHashMap<>();
|
||||
row.put("profile", profile); row.put("maxLoad", cap); row.put("live", live);
|
||||
|
||||
@@ -540,16 +540,66 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
|
||||
* itself is still lost, because there is no OS-level compare-and-swap on a plain file, only this
|
||||
* cooperative narrowing of the gap.
|
||||
*
|
||||
* <p><b>fleetd #285: refuses under {@code memberHerdrSocket} rather than writing somewhere the
|
||||
* member cannot read.</b> Under {@code memberHerdrSocket:} the member pane runs as a
|
||||
* <em>different OS user with its own {@code $HOME}</em> — the same reason {@link
|
||||
* #writeCharterFile} routes the role/reply charter under {@code worktreeRoot} instead of
|
||||
* {@code java.io.tmpdir} and refuses the spawn when it cannot. This method has no equivalent
|
||||
* relocation available: unlike the charter (fleetd's own content, free to place anywhere and
|
||||
* hand to the peer via an argv flag), {@code .claude.json} is a file Claude Code looks up for
|
||||
* ITSELF at a fixed location — {@code CLAUDE_CONFIG_DIR/.claude.json}, or else the member OS
|
||||
* user's own {@code ~/.claude.json}, a path fleetd has no channel to learn. So when {@code
|
||||
* configDir} is unset, there is no member-readable target to seed at all — writing the
|
||||
* unqualified default would land in <em>fleetd's own</em> {@code ~/.claude.json} instead, the
|
||||
* exact defect this fix closes, not a workable fallback. And even with {@code configDir} set,
|
||||
* the file this method itself just wrote is {@code 0600} (owner-only — see {@link
|
||||
* #copyPosixPermissionsIfPresent}), unreadable by a different-uid member unless shared with
|
||||
* {@code worktreeGroup}, the same group {@link EnvAllowListScrub#shareWithGroup} already uses
|
||||
* for the ZDOTDIR scrub (fleetd #213) and the charter file (fleetd #219/#222). So under {@code
|
||||
* memberHerdrSocket} this method requires BOTH {@code configDir} and {@code worktreeGroup}
|
||||
* before it ever touches a file, and refuses the spawn — naming exactly which one is missing —
|
||||
* rather than silently corrupt fleetd's own home or hand the member an unreadable path. This
|
||||
* mirrors {@link #writeCharterFile}'s "refuse, don't degrade" decision: a member spawned without
|
||||
* a readable trust seed is not degraded, it sits on the interactive dialog forever and never
|
||||
* calls {@code fleet_reply} — exactly the failure fleetd #149 exists to prevent, so trading it
|
||||
* for "spawn something" is not worth it. When {@code configDir} and {@code worktreeGroup} are
|
||||
* both present, the write proceeds exactly as below and the resulting file is additionally
|
||||
* chgrp'd/chmod'd group-readable ({@code rw-r-----}) via {@link
|
||||
* EnvAllowListScrub#shareFileWithGroup(Path, String)} so the member's OS user can actually
|
||||
* open it — the
|
||||
* directory itself (unlike {@code worktreeRoot} or the charter's per-spawn directory) is not
|
||||
* fleetd-managed, so its own traversal permissions remain the operator's setup, same as they
|
||||
* already must be for the member to read anything else fleetd points {@code CLAUDE_CONFIG_DIR}
|
||||
* at. With {@code memberHerdrSocket} ABSENT (today's only live mode) every branch below is
|
||||
* byte-identical to before this fix.
|
||||
*
|
||||
* @param configDir the profile's {@code CLAUDE_CONFIG_DIR} ({@code cfg.configDir()}), or
|
||||
* {@code null}/blank to target the default {@code ~/.claude.json}
|
||||
* {@code null}/blank to target the default {@code ~/.claude.json} — refused
|
||||
* outright when {@code memberHerdrSocket} is configured, see above
|
||||
* @param cwd the spawn's resolved working directory — the exact key Claude Code will look
|
||||
* up for itself once it starts there
|
||||
* @throws IllegalStateException when {@code memberHerdrSocket} is configured but {@code
|
||||
* configDir} and/or {@code worktreeGroup} is not — the same
|
||||
* refusal shape as {@link #writeCharterFile}
|
||||
*/
|
||||
private static void seedTrustDialog(String configDir, String cwd) {
|
||||
private void seedTrustDialog(String configDir, String cwd) {
|
||||
if (!isProvisionedWorktree(cwd)) {
|
||||
return;
|
||||
}
|
||||
boolean unsetConfigDir = configDir == null || configDir.isBlank();
|
||||
boolean memberHerdrSocket = memberHerdrSocketConfigured();
|
||||
String group = memberHerdrSocket ? memberGroup() : null;
|
||||
if (memberHerdrSocket && (unsetConfigDir || group == null)) {
|
||||
throw new IllegalStateException("memberHerdrSocket is configured, so the workspace-trust "
|
||||
+ "seed (.claude.json, which gates Claude Code's interactive trust dialog) must be "
|
||||
+ "placed where the member's OS user can read it — configDir, shared via "
|
||||
+ "worktreeGroup — but " + (unsetConfigDir ? "configDir" : "worktreeGroup")
|
||||
+ " is not configured. Refusing to spawn rather than write fleetd's own default "
|
||||
+ "'~/.claude.json' or hand the member a config file it cannot read: that member "
|
||||
+ "would sit on the interactive trust dialog forever and never reach an "
|
||||
+ "injectable state. Configure configDir on this profile and worktreeGroup on the "
|
||||
+ "fleet to enable claude-code member spawns under memberHerdrSocket.");
|
||||
}
|
||||
Path target = unsetConfigDir
|
||||
? Path.of(System.getProperty("user.home"), ".claude.json")
|
||||
: Path.of(configDir, ".claude.json");
|
||||
@@ -559,18 +609,20 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
|
||||
// profile that simply forgot to set configDir gets no signal at all short of the
|
||||
// operator noticing their own file changing. Say so loudly, every time it is about to
|
||||
// happen, rather than only once ever: each occurrence is a live write to a real
|
||||
// person's home config and deserves its own log line.
|
||||
// person's home config and deserves its own log line. (Reached only when
|
||||
// memberHerdrSocket is absent — the block above already refused otherwise.)
|
||||
log.warn("seedTrustDialog: profile has no configDir set, so the workspace-trust seed "
|
||||
+ "for cwd '{}' is about to write the operator's own default '{}' — set "
|
||||
+ "configDir on this profile to target a per-member config file instead",
|
||||
cwd, target);
|
||||
}
|
||||
synchronized (TRUST_JSON_LOCK) {
|
||||
boolean written = false;
|
||||
try {
|
||||
if (target.getParent() != null) {
|
||||
Files.createDirectories(target.getParent());
|
||||
}
|
||||
for (int attempt = 1; attempt <= MAX_TRUST_JSON_CAS_ATTEMPTS; attempt++) {
|
||||
for (int attempt = 1; attempt <= MAX_TRUST_JSON_CAS_ATTEMPTS && !written; attempt++) {
|
||||
byte[] before = Files.isRegularFile(target) ? Files.readAllBytes(target) : null;
|
||||
ObjectNode root = parseTrustJsonOrEmpty(before);
|
||||
JsonNode projectsNode = root.get("projects");
|
||||
@@ -611,26 +663,39 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
|
||||
continue;
|
||||
}
|
||||
writeAtomically(target, newContent);
|
||||
return;
|
||||
written = true;
|
||||
}
|
||||
if (!written) {
|
||||
// fleetd #247: deliberately do NOT write here. A member that starts without the
|
||||
// seed still starts — it may hit the trust dialog fleetd #149 describes and fail
|
||||
// to reach an injectable state, but that failure is visible (herdr reports it,
|
||||
// the spawn-readiness gate times out) and recoverable (retry the spawn). Writing
|
||||
// our stale copy over whatever the other writer left would be silent and, if that
|
||||
// other writer is the operator's own live session, could destroy real
|
||||
// configuration — fail toward the recoverable outcome, not the silent one.
|
||||
log.warn("seedTrustDialog: gave up seeding workspace-trust for cwd '{}' into '{}' "
|
||||
+ "after {} attempts — another writer (most plausibly the operator's own "
|
||||
+ "live Claude Code sharing this file) kept changing it faster than we "
|
||||
+ "could re-read it, so nothing was written; the member may show the "
|
||||
+ "trust dialog instead", cwd, target, MAX_TRUST_JSON_CAS_ATTEMPTS);
|
||||
}
|
||||
// fleetd #247: deliberately do NOT write here. A member that starts without the
|
||||
// seed still starts — it may hit the trust dialog fleetd #149 describes and fail to
|
||||
// reach an injectable state, but that failure is visible (herdr reports it, the
|
||||
// spawn-readiness gate times out) and recoverable (retry the spawn). Writing our
|
||||
// stale copy over whatever the other writer left would be silent and, if that other
|
||||
// writer is the operator's own live session, could destroy real configuration —
|
||||
// fail toward the recoverable outcome, not the silent one.
|
||||
log.warn("seedTrustDialog: gave up seeding workspace-trust for cwd '{}' into '{}' "
|
||||
+ "after {} attempts — another writer (most plausibly the operator's own "
|
||||
+ "live Claude Code sharing this file) kept changing it faster than we could "
|
||||
+ "re-read it, so nothing was written; the member may show the trust dialog "
|
||||
+ "instead", cwd, target, MAX_TRUST_JSON_CAS_ATTEMPTS);
|
||||
} catch (Exception e) {
|
||||
log.debug("cannot seed workspace-trust entry for cwd '{}' into '{}'", cwd, target, e);
|
||||
return;
|
||||
}
|
||||
// fleetd #285: the write above lands as fleetd's own OS user; under memberHerdrSocket
|
||||
// that is NOT the member's OS user, so without this the member still cannot read the
|
||||
// file it exists to seed — a silent readiness timeout with the write looking "done".
|
||||
// Deliberately OUTSIDE the swallow-all catch above: a group that fails to resolve here
|
||||
// means the seed is unreadable despite a successful write, which must fail as loudly as
|
||||
// writeCharterFile's own EnvAllowListScrub.shareWithGroup call already does.
|
||||
if (written && memberHerdrSocket) {
|
||||
EnvAllowListScrub.shareFileWithGroup(target, group);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/** Bound on {@link #seedTrustDialog}'s fleetd #247 compare-and-swap retry loop. */
|
||||
private static final int MAX_TRUST_JSON_CAS_ATTEMPTS = 5;
|
||||
|
||||
|
||||
@@ -181,6 +181,33 @@ public final class EnvAllowListScrub {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #285: share ONE file with {@code group}, read-only ({@code rw-r-----}) — the same
|
||||
* per-file mode {@link #shareWithGroup} applies to a directory's entries, and the same error
|
||||
* shapes, but without touching a parent directory. Used for a file fleetd writes into a
|
||||
* directory it does NOT own — {@code configDir}'s own traversal permissions stay the
|
||||
* operator's setup — where the directory-wide {@link #shareWithGroup} would be wrong.
|
||||
*
|
||||
* @throws UncheckedIOException when {@code group} does not resolve on this host, the
|
||||
* filesystem has no POSIX group ownership, or a
|
||||
* group-ownership/permission call is refused
|
||||
*/
|
||||
static void shareFileWithGroup(Path file, String group) {
|
||||
try {
|
||||
GroupPrincipal principal = file.getFileSystem().getUserPrincipalLookupService()
|
||||
.lookupPrincipalByGroupName(group);
|
||||
setGroupAndPermissions(file, principal, "rw-r-----");
|
||||
} catch (IOException e) {
|
||||
throw new UncheckedIOException("cannot share generated file " + file + " with group '"
|
||||
+ group + "' — the group must exist, and the fleetd operator ("
|
||||
+ System.getProperty("user.name") + ") must be a member of it", e);
|
||||
} catch (UnsupportedOperationException e) {
|
||||
throw new UncheckedIOException("cannot share generated file " + file + " with group '"
|
||||
+ group + "' — this filesystem does not support POSIX group ownership",
|
||||
new IOException(e));
|
||||
}
|
||||
}
|
||||
|
||||
private static void setGroupAndPermissions(Path path, GroupPrincipal group, String perms) throws IOException {
|
||||
PosixFileAttributeView view = Files.getFileAttributeView(path, PosixFileAttributeView.class);
|
||||
if (view == null) {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -913,9 +926,14 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
* {@link #reapOrphanWorkers() orphan-reap} and spawn-gate-timeout paths, plus any caller that
|
||||
* passes a pane directly, keep working without an owning id.
|
||||
*
|
||||
* <p>Resolves the tab from the pane <em>before</em> closing it. An already-gone pane/tab
|
||||
* (repeated DELETE, crashed peer) is treated as success; any other failure propagates so a
|
||||
* genuinely failed teardown is not reported as done.
|
||||
* <p>Resolves the tab from the pane <em>before</em> closing it. {@code agents.close} (the pane)
|
||||
* is the one step whose failure means the teardown itself may not have happened: an already-gone
|
||||
* pane (repeated DELETE, crashed peer) is treated as success, but any other failure propagates so
|
||||
* a genuinely failed teardown is not reported as done. {@code spaces.closeTab} (fleetd #293) is
|
||||
* different — by the time it runs the pane is already closed, so it is cosmetic workspace tidying
|
||||
* rather than a real teardown failure, and a failure there is logged and never propagates, so it
|
||||
* cannot mask the two cleanups below it ({@link #releaseZdotdir}, and the caller's worktree
|
||||
* removal in {@code SessionManager.release}).
|
||||
*/
|
||||
@Override
|
||||
public void stop(String idOrPane) {
|
||||
@@ -934,7 +952,25 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
log.debug("pane.close({}) ignored — already gone: {}", paneId, e.getMessage());
|
||||
}
|
||||
if (loc != null && loc.tabPaneCount() == 1) {
|
||||
spaces.closeTab(loc.tabId());
|
||||
// fleetd #293: the pane above is already closed by this point, so a failing tab.close is
|
||||
// cosmetic workspace tidying, not a real teardown failure — it must not mask the two
|
||||
// cleanups below it (releaseZdotdir, and the caller's worktree removal). Unlike
|
||||
// agents.close above, this is not narrowed to "already gone": any failure here, whatever
|
||||
// its cause, is one we continue past, so we log it at WARN (not debug) with the tab id a
|
||||
// person can go close by hand.
|
||||
try {
|
||||
spaces.closeTab(loc.tabId());
|
||||
} catch (RuntimeException e) {
|
||||
// Caught as RuntimeException, not HerdrException, to match releaseZdotdir's own
|
||||
// guard five lines below. Today the two are the same set — HerdrCodec wraps every
|
||||
// encode/decode failure and UnixSocketHerdrClient wraps every IOException, so
|
||||
// HerdrException is all closeTab can actually throw. Narrowing to it anyway would
|
||||
// leave this step guarded against the expected failure and bare against any other,
|
||||
// which is the exact asymmetry fleetd #293 exists to remove. No behaviour change
|
||||
// today; it stops a later change inside WorkspaceControl.closeTab reopening it.
|
||||
log.warn("tab.close({}) failed — the pane is already torn down, so continuing; the "
|
||||
+ "tab may need manual cleanup: {}", loc.tabId(), e.getMessage());
|
||||
}
|
||||
} else if (loc != null) {
|
||||
log.debug("not closing tab {} — it holds {} panes (not a dedicated peer tab)",
|
||||
loc.tabId(), loc.tabPaneCount());
|
||||
@@ -973,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();
|
||||
@@ -987,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)) {
|
||||
|
||||
@@ -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
|
||||
@@ -927,7 +945,16 @@ public final class MessageService {
|
||||
}
|
||||
try {
|
||||
CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(workerSession);
|
||||
// #282: mirror send()'s registration (:802) so a SECOND fleet_ask inside this same
|
||||
// resumed turn can re-associate the async ticket with its new turnId via
|
||||
// markAsyncQuestion — without this, that second ask has no Task to attach to, and
|
||||
// markAsyncQuestion silently returns null.
|
||||
Task task = asyncTasksByTurn.get(turnId);
|
||||
if (task != null) {
|
||||
asyncTasksByWaiter.put(reply, task);
|
||||
}
|
||||
if (!rendezvous.answerAsk(turnId, content)) {
|
||||
asyncTasksByWaiter.remove(reply);
|
||||
rendezvous.close(workerSession, reply);
|
||||
return new Reply(Outcome.STALE_TURN, null); // lapsed between the lookup and the unblock
|
||||
}
|
||||
@@ -935,7 +962,21 @@ public final class MessageService {
|
||||
try {
|
||||
Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
|
||||
Reply result = new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
|
||||
finishAsyncTask(turnId, result);
|
||||
// #282: this waiter can resolve with a FRESH question rather than a terminal reply —
|
||||
// the worker chained a second fleet_ask before replying. Mirror sendAsync's own guard
|
||||
// (:1000) and leave the ticket open (markAsyncQuestion above already re-armed it under
|
||||
// the new turnId) instead of completing it here with a QUESTION "reply".
|
||||
// Measured when #282 was merged: this guard is DEFENCE IN DEPTH, not the thing
|
||||
// that makes the chained ask work. ask() calls markAsyncQuestion (:860) before
|
||||
// resolveQuestion (:861), so by the time this thread wakes, the task has already
|
||||
// moved to the new turnId and finishAsyncTask(oldTurnId, ...) finds nothing. Removing
|
||||
// this guard alone leaves the test green. Keep it anyway: it mirrors sendAsync's
|
||||
// sibling guard, and that sibling's own comment (:1017) warns the two orderings are
|
||||
// not something to rely on. Do NOT delete it as dead code without re-checking that
|
||||
// ordering, and do not treat it as the sole protection either.
|
||||
if (result.outcome() != Outcome.QUESTION) {
|
||||
finishAsyncTask(turnId, result);
|
||||
}
|
||||
return result;
|
||||
} catch (TimeoutException e) {
|
||||
// The worker resumed but hasn't replied yet — no completion fallback arms an answered
|
||||
@@ -948,6 +989,7 @@ public final class MessageService {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new IllegalStateException("interrupted awaiting reply from " + workerSession, e);
|
||||
} finally {
|
||||
asyncTasksByWaiter.remove(reply);
|
||||
rendezvous.close(workerSession, reply);
|
||||
}
|
||||
} finally {
|
||||
@@ -1106,9 +1148,16 @@ public final class MessageService {
|
||||
private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
|
||||
Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
|
||||
if (task != null) {
|
||||
String previousTurnId = task.turnId;
|
||||
task.question = new Reply(Outcome.QUESTION, text, turnId);
|
||||
task.turnId = turnId;
|
||||
asyncTasksByTurn.put(turnId, task);
|
||||
// #282: a second fleet_ask in the same resumed turn re-arms an already-answered task
|
||||
// (answer() re-registers it in asyncTasksByWaiter) under a FRESH turnId — drop the old
|
||||
// key so asyncTasksByTurn does not keep growing by one stale entry per chained ask.
|
||||
if (previousTurnId != null && !previousTurnId.equals(turnId)) {
|
||||
asyncTasksByTurn.remove(previousTurnId, task);
|
||||
}
|
||||
}
|
||||
return task;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
@@ -47,6 +48,22 @@ import java.util.stream.Collectors;
|
||||
*/
|
||||
public final class FleetApp {
|
||||
|
||||
/** The authorization action the matching route handler hands to {@link #allow}. */
|
||||
static Authz.Action routeAction(String route) {
|
||||
return switch (route) {
|
||||
case "GET /metrics" -> Authz.Action.METRICS;
|
||||
case "POST /members" -> Authz.Action.SPAWN;
|
||||
case "DELETE /members/{paneId}" -> Authz.Action.STOP;
|
||||
case "POST /sessions/{id}/message" -> Authz.Action.SEND;
|
||||
case "POST /sessions/{id}/reply" -> Authz.Action.REPLY;
|
||||
case "GET /sessions/{id}/replies" -> Authz.Action.DRAIN;
|
||||
case "POST /sessions/{id}/ask" -> Authz.Action.ASK;
|
||||
case "GET /sessions", "GET /agents", "GET /members", "GET /profiles",
|
||||
"GET /member-credentials", "GET /sessions/{id}/status", "GET /tasks/{ticket}" -> Authz.Action.READ;
|
||||
default -> throw new IllegalArgumentException("route has no authorization gate: " + route);
|
||||
};
|
||||
}
|
||||
|
||||
/** Default blocking window for a message; kept under typical HTTP idle timeouts. */
|
||||
private static final long DEFAULT_MESSAGE_TIMEOUT_MS = 25_000;
|
||||
private static final long MAX_MESSAGE_TIMEOUT_MS = 120_000;
|
||||
@@ -70,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();
|
||||
|
||||
/**
|
||||
@@ -130,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;
|
||||
@@ -140,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. */
|
||||
@@ -222,7 +264,7 @@ public final class FleetApp {
|
||||
|
||||
/** Prometheus scrape endpoint (CB-502). */
|
||||
private void metrics(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.METRICS, null)) {
|
||||
if (!allow(ctx, routeAction("GET /metrics"), null)) {
|
||||
return;
|
||||
}
|
||||
ctx.status(200).contentType("text/plain; version=0.0.4; charset=utf-8").result(metrics.render());
|
||||
@@ -294,7 +336,7 @@ public final class FleetApp {
|
||||
* member workspace (they live on the member daemon only).
|
||||
*/
|
||||
private void sessions(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.READ, null)) {
|
||||
if (!allow(ctx, routeAction("GET /sessions"), null)) {
|
||||
return;
|
||||
}
|
||||
List<Map<String, Object>> out = new ArrayList<>();
|
||||
@@ -319,52 +361,76 @@ public final class FleetApp {
|
||||
|
||||
/** Discovery: every agent herdr tracks, keyed by its Claude session UUID. */
|
||||
private void agents(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.READ, null)) {
|
||||
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. */
|
||||
private void listMembers(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.READ, null)) {
|
||||
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, Authz.Action.READ, null)) {
|
||||
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));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -376,7 +442,7 @@ public final class FleetApp {
|
||||
* name list, which is exactly what let the list drift silently behind the real policy.
|
||||
*/
|
||||
private void memberCredentials(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.READ, null)) {
|
||||
if (!allow(ctx, routeAction("GET /member-credentials"), null)) {
|
||||
return;
|
||||
}
|
||||
MemberCredentialPolicyView view = memberCredentials.get();
|
||||
@@ -396,7 +462,7 @@ public final class FleetApp {
|
||||
* the subscription boundary, 400 for an unknown profile.
|
||||
*/
|
||||
private void spawnMember(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.SPAWN, null)) {
|
||||
if (!allow(ctx, routeAction("POST /members"), null)) {
|
||||
return;
|
||||
}
|
||||
String role = ctx.queryParam("role");
|
||||
@@ -467,7 +533,7 @@ public final class FleetApp {
|
||||
/** Tear a worker down by pane id. */
|
||||
private void stopMember(Context ctx) {
|
||||
String paneId = ctx.pathParam("paneId");
|
||||
if (!allow(ctx, Authz.Action.STOP, paneId)) {
|
||||
if (!allow(ctx, routeAction("DELETE /members/{paneId}"), paneId)) {
|
||||
return;
|
||||
}
|
||||
sessions.release(paneId);
|
||||
@@ -482,7 +548,7 @@ public final class FleetApp {
|
||||
*/
|
||||
private void sendMessage(Context ctx) {
|
||||
String id = ctx.pathParam("id");
|
||||
if (!allow(ctx, Authz.Action.SEND, id)) {
|
||||
if (!allow(ctx, routeAction("POST /sessions/{id}/message"), id)) {
|
||||
return;
|
||||
}
|
||||
String content;
|
||||
@@ -569,7 +635,7 @@ public final class FleetApp {
|
||||
*/
|
||||
private void askMessage(Context ctx) {
|
||||
String id = ctx.pathParam("id");
|
||||
if (!allow(ctx, Authz.Action.ASK, id)) {
|
||||
if (!allow(ctx, routeAction("POST /sessions/{id}/ask"), id)) {
|
||||
return;
|
||||
}
|
||||
String question;
|
||||
@@ -608,7 +674,7 @@ public final class FleetApp {
|
||||
// The rule that matters: a worker may reply only as itself. Over MCP this was already true
|
||||
// structurally (identity comes from the connection, never an argument); over REST the path
|
||||
// id was simply trusted, so this is where the invariant actually gets enforced.
|
||||
if (!allow(ctx, Authz.Action.REPLY, id)) {
|
||||
if (!allow(ctx, routeAction("POST /sessions/{id}/reply"), id)) {
|
||||
return;
|
||||
}
|
||||
String content;
|
||||
@@ -618,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));
|
||||
}
|
||||
|
||||
@@ -629,7 +707,7 @@ public final class FleetApp {
|
||||
*/
|
||||
private void drainReplies(Context ctx) {
|
||||
String id = ctx.pathParam("id");
|
||||
if (!allow(ctx, Authz.Action.DRAIN, id)) {
|
||||
if (!allow(ctx, routeAction("GET /sessions/{id}/replies"), id)) {
|
||||
return;
|
||||
}
|
||||
var replies = messages.drainReplies(id);
|
||||
@@ -647,7 +725,7 @@ public final class FleetApp {
|
||||
*/
|
||||
private void sessionStatus(Context ctx) {
|
||||
String id = ctx.pathParam("id");
|
||||
if (!allow(ctx, Authz.Action.READ, id)) {
|
||||
if (!allow(ctx, routeAction("GET /sessions/{id}/status"), id)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
@@ -672,7 +750,7 @@ public final class FleetApp {
|
||||
|
||||
/** Poll an async (wait:false) delegation by ticket. 404 for an unknown/expired ticket. */
|
||||
private void taskStatus(Context ctx) {
|
||||
if (!allow(ctx, Authz.Action.READ, null)) {
|
||||
if (!allow(ctx, routeAction("GET /tasks/{ticket}"), null)) {
|
||||
return;
|
||||
}
|
||||
MessageService.TaskView v = messages.poll(ctx.pathParam("ticket"));
|
||||
|
||||
@@ -213,13 +213,18 @@ public final class GitWorktrees implements Worktrees {
|
||||
worktreePath, cleanup.getMessage());
|
||||
}
|
||||
try {
|
||||
exec("git", "-C", repoRoot, "branch", "-D", branch);
|
||||
deleteBranch(repoRoot, branch);
|
||||
} catch (RuntimeException cleanup) {
|
||||
log.warn("failed to remove leaked branch {} after provisioning error: {}",
|
||||
branch, cleanup.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteBranch(String repoRoot, String branch) {
|
||||
exec("git", "-C", repoRoot, "branch", "-D", branch);
|
||||
}
|
||||
|
||||
/**
|
||||
* A linked worktree shares its primary checkout's git config. Remove HTTPS user info before
|
||||
* adding one, so a credential accidentally embedded in that config cannot reach the member.
|
||||
|
||||
@@ -334,7 +334,20 @@ public final class SessionManager implements TurnListener {
|
||||
// slot that no longer appears in the roster and can never be reclaimed.
|
||||
launcher.stop(paneId);
|
||||
if (removed != null && !preserveWorktree && removed.worktree() != null) {
|
||||
worktrees.remove(worktrees.repoRoot(removed.cwd()), removed.worktree());
|
||||
// fleetd #283: this is the one cleanup step in this method that used to be bare. By the
|
||||
// time it runs, the registry entry, the retained handle, and the pane are all already
|
||||
// gone — so a throw here (a stale index lock, a slow filesystem, `remove`'s own 30s exec
|
||||
// timeout) must not escape release(): there is no retry path (a second stop on this
|
||||
// paneId is a no-op), and the caller would otherwise see a "failed stop" for a session
|
||||
// that is in fact fully torn down. Log and swallow, matching every sibling step above.
|
||||
try {
|
||||
worktrees.remove(worktrees.repoRoot(removed.cwd()), removed.worktree());
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("failed to remove worktree {} for pane={} terminal={} after release: the "
|
||||
+ "pane is already stopped and the session already deregistered, so this is "
|
||||
+ "not retryable — the directory must be reclaimed manually: {}",
|
||||
removed.worktree(), paneId, removed.terminalId(), e.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -504,11 +517,26 @@ public final class SessionManager implements TurnListener {
|
||||
log.warn("spawn failed for profile={} role={} branch={} path={}: {}",
|
||||
preResolvedProfile, memberRole, branch, path, e.getMessage());
|
||||
if (path != null) {
|
||||
// fleetd #283: this catch covers every failure AFTER worktrees.add() returned —
|
||||
// overlayParity, shareWithGroup, launcher.spawn itself — so by this point `branch`
|
||||
// was actually created in git. #274 fixed the sibling failure INSIDE add() by having
|
||||
// GitWorktrees.cleanupAfterAddFailure delete both the worktree and the branch it
|
||||
// provisioned; this path removed only the worktree and left the branch orphaned. A
|
||||
// spawn failure here is routine (a quarantined credential, a backend refusal), so
|
||||
// every occurrence leaked a `worker/<slug>-<nonce>` branch nothing ever pointed at
|
||||
// again. Reuse the same Worktrees.deleteBranch GitWorktrees already has, rather than
|
||||
// a second copy of the git command. Best-effort and log-only, like the worktree
|
||||
// removal right above it — neither cleanup step may mask the original exception.
|
||||
try {
|
||||
worktrees.remove(repoRoot, path);
|
||||
} catch (RuntimeException cleanup) {
|
||||
log.warn("failed to clean up worktree {} after spawn error: {}", path, cleanup.getMessage());
|
||||
}
|
||||
try {
|
||||
worktrees.deleteBranch(repoRoot, branch);
|
||||
} catch (RuntimeException cleanup) {
|
||||
log.warn("failed to clean up branch {} after spawn error: {}", branch, cleanup.getMessage());
|
||||
}
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
@@ -836,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
|
||||
|
||||
@@ -11,6 +11,15 @@ public interface Worktrees {
|
||||
/** git -C <repoRoot> worktree remove --force <path>. Idempotent (already-gone tolerated). */
|
||||
void remove(String repoRoot, String worktreePath);
|
||||
|
||||
/**
|
||||
* git -C {@code repoRoot} branch -D {@code branch}. Force-deletes a branch that has no other
|
||||
* owner — used only on the failed-provisioning path (fleetd #274, #283), never on a normal
|
||||
* release: {@link SessionManager#release} deliberately leaves a released session's branch
|
||||
* behind so a lead can still recover the work, and this method must never be called from
|
||||
* that path.
|
||||
*/
|
||||
void deleteBranch(String repoRoot, String branch);
|
||||
|
||||
/**
|
||||
* True when the worktree holds uncommitted changes the bridge cannot see: tracked
|
||||
* modifications, staged files, or untracked files. {@code git status --porcelain} is the
|
||||
|
||||
@@ -226,4 +226,44 @@ class FleetdBackendErrorSinkTest {
|
||||
assertTrue(remaining.isPresent(), "two distinct targets must start a cool-off");
|
||||
assertEquals(1, leadClient.sendCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a backend-error session no longer blocks the real maxLoad spawn gate")
|
||||
void backendErrorSessionDoesNotBlockFreshSpawnAtMaxLoad() {
|
||||
SessionManager sessions = capacityLimitedSessions();
|
||||
MemberSession failed = sessions.acquire("terra", null, null, null);
|
||||
|
||||
assertTrue(sessions.onBackendError(failed.terminalId(), "backend exited"));
|
||||
|
||||
MemberSession fresh = sessions.acquire("terra", null, null, null);
|
||||
assertEquals("terra", fresh.profile(), "the real maxLoad gate grants a fresh spawn after a backend error");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a failed session no longer blocks the real maxLoad spawn gate")
|
||||
void failedSessionDoesNotBlockFreshSpawnAtMaxLoad() {
|
||||
SessionManager sessions = capacityLimitedSessions();
|
||||
MemberSession failed = sessions.acquire("terra", null, null, null);
|
||||
|
||||
sessions.onTurnFailed(failed.terminalId());
|
||||
|
||||
MemberSession fresh = sessions.acquire("terra", null, null, null);
|
||||
assertEquals("terra", fresh.profile(), "the real maxLoad gate grants a fresh spawn after a failed turn");
|
||||
}
|
||||
|
||||
private static SessionManager capacityLimitedSessions() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FleetConfig.Profile profile = new FleetConfig.Profile("terra", "http://gx00.gw:8000", "coder",
|
||||
null, "FLEETD_WORKER_TOKEN", List.of("claude"), "tab", "fleetd-workers",
|
||||
"w #{n}", null, null, null, null, null, null, null, 1.0f, 1);
|
||||
Map<String, FleetConfig.Profile> profiles = Map.of("terra", profile);
|
||||
ClaudeCodeLauncher adapter = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")), profiles, "terra", _ -> "tok");
|
||||
AtomicReference<SessionManager> sessionsRef = new AtomicReference<>();
|
||||
CompositePeerLauncher workers = new CompositePeerLauncher(List.of(adapter), "terra", profiles,
|
||||
PlacementPolicies.fixed(), name -> Fleetd.liveSessionCount(sessionsRef.get().roster(), name));
|
||||
SessionManager sessions = new SessionManager(workers);
|
||||
sessionsRef.set(sessions);
|
||||
return sessions;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,7 +24,9 @@ import org.slf4j.LoggerFactory;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
@@ -434,4 +436,120 @@ class FleetHealthMonitorTest {
|
||||
logger.detachAppender(appender);
|
||||
}
|
||||
}
|
||||
|
||||
// --- fleetd #280: a GONE/NEVER_READY guess whose fleet_ask lapses AFTER the first sweep must
|
||||
// still be swept, without ever reaching into a target that has since recovered or left the
|
||||
// roster. See FleetHealthMonitor.recheckTerminalTarget's javadoc for the full reachability chain.
|
||||
|
||||
@Test void terminalTransitionSchedulesExactlyOneDelayedRecheck() {
|
||||
ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1);
|
||||
FleetHealthMonitor monitor = monitor(new FakeHerdr(),
|
||||
List.of(member("term_a", MemberSession.State.BUSY, 0, 0)), scheduler, () -> 1, 600,
|
||||
(_, _) -> { });
|
||||
|
||||
monitor.reportTransition("term_a", HealthState.GONE);
|
||||
|
||||
// Driven via reportTransition directly (not tick()), so the queue holds only the recheck.
|
||||
assertEquals(1, scheduler.getQueue().size());
|
||||
ScheduledFuture<?> scheduled = (ScheduledFuture<?>) scheduler.getQueue().peek();
|
||||
assertTrue(scheduled.getDelay(TimeUnit.SECONDS) > 100,
|
||||
"the delay must clear the worst-case fleet_ask lapse window (up to 115s)");
|
||||
|
||||
// An unchanged tick must not queue a second one (CB-580's fire-once rule extends to this).
|
||||
monitor.reportTransition("term_a", HealthState.GONE);
|
||||
assertEquals(1, scheduler.getQueue().size());
|
||||
monitor.stop();
|
||||
}
|
||||
|
||||
@Test void recheckIsANoOpOnceTheTargetHasRecovered() {
|
||||
RecordingFailTarget failTarget = new RecordingFailTarget();
|
||||
FleetHealthMonitor monitor = monitorWith(failTarget);
|
||||
monitor.reportTransition("term_a", HealthState.GONE);
|
||||
assertEquals(1, failTarget.calls.size());
|
||||
|
||||
monitor.reportTransition("term_a", HealthState.IDLE); // recovered before the recheck fired
|
||||
monitor.recheckTerminalTarget("term_a", HealthState.GONE);
|
||||
|
||||
assertEquals(1, failTarget.calls.size(), "a recovered target must not be reached into again");
|
||||
monitor.stop();
|
||||
}
|
||||
|
||||
@Test void recheckIsANoOpForATargetItNeverObserved() {
|
||||
// Mirrors "left the roster": tick() prunes states.keySet() to the current roster on release
|
||||
// (see FleetHealthMonitor.tick), so a target this monitor never recorded is the same case.
|
||||
RecordingFailTarget failTarget = new RecordingFailTarget();
|
||||
FleetHealthMonitor monitor = monitorWith(failTarget);
|
||||
|
||||
monitor.recheckTerminalTarget("term_never_seen", HealthState.GONE);
|
||||
|
||||
assertEquals(0, failTarget.calls.size(), "an untracked/released target must not be reached into");
|
||||
monitor.stop();
|
||||
}
|
||||
|
||||
/**
|
||||
* The scenario from the ticket, end to end, driven through the real {@link MessageService}: a
|
||||
* target's ask is still genuinely open when health first observes GONE (sweep must skip it,
|
||||
* exactly as {@code abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer} pins), the ask then lapses
|
||||
* on its own, an unchanged tick still must not refire, and only the delayed recheck sweeps the
|
||||
* now-lapsed ticket to FAILED.
|
||||
*/
|
||||
@Test void delayedRecheckSweepsATicketWhoseAskLapsedAfterGoneWasFirstObserved() throws Exception {
|
||||
FakeHerdr herdr = new FakeHerdr().withAgent("worker", "term_a", "pane-term_a", "tab_a")
|
||||
.readText("$ prompt");
|
||||
AgentControl agents = new AgentControl(herdr);
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
Injector injector = new Injector(agents);
|
||||
InMemoryReplyInbox inbox = new InMemoryReplyInbox();
|
||||
inbox.own("term_a");
|
||||
MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
|
||||
FleetHealthMonitor monitor = new FleetHealthMonitor(agents,
|
||||
() -> List.of(member("term_a", MemberSession.State.READY, 0, 0)), messages,
|
||||
new ScheduledThreadPoolExecutor(1), () -> 1, 60, 600, messages::abandon);
|
||||
|
||||
String ticket = messages.sendAsync("term_a", "task that asks");
|
||||
long deadline = System.currentTimeMillis() + 2000;
|
||||
while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
|
||||
Thread.sleep(5);
|
||||
}
|
||||
assertTrue(rendezvous.isWaiting("term_a"), "the async send should have opened its waiter");
|
||||
injector.onStatus("term_a", AgentStatus.IDLE); // deliver the task
|
||||
injector.onStatus("term_a", AgentStatus.WORKING); // the worker picks it up
|
||||
|
||||
// The worker asks, with a short timeout so its own fleet_ask lapses quickly in test time.
|
||||
CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
|
||||
() -> messages.ask("term_a", "which config?", 200));
|
||||
awaitPhase(messages, ticket, MessageService.Phase.ASKING);
|
||||
|
||||
monitor.reportTransition("term_a", HealthState.GONE);
|
||||
assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase(),
|
||||
"the first sweep must not fail a ticket that is still genuinely being asked");
|
||||
|
||||
// The worker's own fleet_ask now lapses on its own — task.question clears to null.
|
||||
assertEquals(MessageService.AskOutcome.TIMED_OUT, ask.get(5, TimeUnit.SECONDS).outcome());
|
||||
|
||||
// An unchanged tick still must not refire (CB-580).
|
||||
monitor.reportTransition("term_a", HealthState.GONE);
|
||||
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase());
|
||||
|
||||
// The delayed recheck scheduled for the original transition finally sweeps it.
|
||||
monitor.recheckTerminalTarget("term_a", HealthState.GONE);
|
||||
assertEquals(MessageService.Phase.FAILED, messages.poll(ticket).phase());
|
||||
|
||||
monitor.stop();
|
||||
}
|
||||
|
||||
private static MessageService.TaskView awaitPhase(MessageService messages, String ticket,
|
||||
MessageService.Phase phase) throws InterruptedException {
|
||||
long deadline = System.currentTimeMillis() + 2000;
|
||||
MessageService.TaskView view;
|
||||
do {
|
||||
view = messages.poll(ticket);
|
||||
if (view.phase() == phase) {
|
||||
return view;
|
||||
}
|
||||
Thread.sleep(5);
|
||||
} while (System.currentTimeMillis() < deadline);
|
||||
assertEquals(phase, view.phase());
|
||||
return view;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
/**
|
||||
@@ -39,8 +40,12 @@ 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<>();
|
||||
private String tabCloseErrorCode = null;
|
||||
private final Map<String, String> tabCloseErrorCodeFor = new ConcurrentHashMap<>();
|
||||
private String agentSendErrorCode = null;
|
||||
private boolean noPanes = false;
|
||||
private volatile String agentStatus = "idle"; // steady-state agent.get status
|
||||
@@ -80,18 +85,56 @@ 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;
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make {@code pane.close} fail with this herdr error code. */
|
||||
/** Make {@code pane.close} fail with this herdr error code, for every pane. */
|
||||
public FakeHerdr paneCloseFailsWith(String code) {
|
||||
this.paneCloseErrorCode = code;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make {@code pane.close} fail with this herdr error code, but only for the given {@code
|
||||
* pane_id} — every other pane's {@code pane.close} still succeeds. Unlike {@link
|
||||
* #paneCloseFailsWith}, which fails every call regardless of which pane it targets, this lets a
|
||||
* test reap/release several sessions at once and make exactly one of them fail to stop, so the
|
||||
* others' teardown can be asserted to proceed normally (fleetd #290).
|
||||
*/
|
||||
public FakeHerdr paneCloseFailsForPane(String paneId, String code) {
|
||||
this.paneCloseErrorCodeFor.put(paneId, code);
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make {@code tab.close} fail with this herdr error code, for every tab. */
|
||||
public FakeHerdr tabCloseFailsWith(String code) {
|
||||
this.tabCloseErrorCode = code;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make {@code tab.close} fail with this herdr error code, but only for the given {@code
|
||||
* tab_id} — every other tab's {@code tab.close} still succeeds. The {@code tab.close}
|
||||
* counterpart to {@link #paneCloseFailsForPane} (fleetd #290): lets a test make exactly one
|
||||
* session's tab teardown fail while proving the rest of {@code stop()} — {@code
|
||||
* releaseZdotdir}, and the caller's worktree removal — still runs (fleetd #293). Named "ForTab"
|
||||
* rather than "ForPane" (unlike its sibling) because {@code tab.close} keys on {@code tab_id},
|
||||
* not a pane id.
|
||||
*/
|
||||
public FakeHerdr tabCloseFailsForTab(String tabId, String code) {
|
||||
this.tabCloseErrorCodeFor.put(tabId, code);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make {@code pane.list} report no panes at all — models a second herdr daemon (CB-185) that
|
||||
* simply does not host the pane a {@link PaneLocator} is searching for.
|
||||
@@ -275,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(
|
||||
@@ -338,7 +385,17 @@ public final class FakeHerdr implements HerdrClient {
|
||||
.formatted(workerTabPaneCount,
|
||||
seeded.isEmpty() ? "" : "," + String.join(",", seeded)));
|
||||
}
|
||||
case "tab.close" -> mapper.readTree("{\"type\":\"ok\"}");
|
||||
case "tab.close" -> {
|
||||
Object tabIdParam = params instanceof Map<?, ?> m ? m.get("tab_id") : null;
|
||||
String perTabCode = tabIdParam == null ? null
|
||||
: tabCloseErrorCodeFor.get(String.valueOf(tabIdParam));
|
||||
String code = perTabCode != null ? perTabCode : tabCloseErrorCode;
|
||||
if (code != null) {
|
||||
throw new HerdrException("herdr error [" + code + "]: tab.close failed",
|
||||
code, null);
|
||||
}
|
||||
yield mapper.readTree("{\"type\":\"ok\"}");
|
||||
}
|
||||
case "pane.get" -> mapper.readTree("""
|
||||
{"type":"pane_info","pane":{"pane_id":"w9:pW","workspace_id":"w9",
|
||||
"tab_id":"w9:t2","agent_status":"idle"}}""");
|
||||
@@ -360,9 +417,13 @@ public final class FakeHerdr implements HerdrClient {
|
||||
"foreground_processes":[]}}""");
|
||||
}
|
||||
case "pane.close" -> {
|
||||
if (paneCloseErrorCode != null) {
|
||||
throw new HerdrException("herdr error [" + paneCloseErrorCode + "]: pane.close failed",
|
||||
paneCloseErrorCode, null);
|
||||
Object paneIdParam = params instanceof Map<?, ?> m ? m.get("pane_id") : null;
|
||||
String perPaneCode = paneIdParam == null ? null
|
||||
: paneCloseErrorCodeFor.get(String.valueOf(paneIdParam));
|
||||
String code = perPaneCode != null ? perPaneCode : paneCloseErrorCode;
|
||||
if (code != null) {
|
||||
throw new HerdrException("herdr error [" + code + "]: pane.close failed",
|
||||
code, null);
|
||||
}
|
||||
yield mapper.readTree("{\"type\":\"ok\"}");
|
||||
}
|
||||
|
||||
@@ -24,8 +24,13 @@ import io.modelcontextprotocol.spec.McpSchema;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
@@ -44,6 +49,9 @@ import static org.junit.jupiter.api.Assertions.*;
|
||||
*/
|
||||
class FleetMcpAuthzTest {
|
||||
|
||||
private static final Path MCP_SOURCE = Path.of("src/main/java/dev/ltms/fleet/mcp/FleetMcp.java");
|
||||
private static final Pattern TOOL_REGISTRATION = Pattern.compile("tool\\(\\\"(fleet_[a-z_]+)\\\"");
|
||||
|
||||
private final FakeHerdr herdr = new FakeHerdr();
|
||||
private final AgentControl agents = new AgentControl(herdr);
|
||||
private Metrics metrics;
|
||||
@@ -201,6 +209,42 @@ class FleetMcpAuthzTest {
|
||||
"a blank target is an absent target");
|
||||
}
|
||||
|
||||
@Test
|
||||
void everyRegisteredToolHasItsHandlerActionPinned() {
|
||||
Set<String> registered = toolsTheServerRegisters();
|
||||
assertTrue(registered.size() >= 10,
|
||||
"scraped only " + registered.size() + " tool registrations from FleetMcp (" + registered
|
||||
+ "); the server registers eleven, so the tool(\"…\") scrape has stopped matching");
|
||||
registered.forEach(tool -> assertDoesNotThrow(() -> FleetMcp.toolAction(tool, Map.of()),
|
||||
() -> tool + " is registered but has no pinned authorization action"));
|
||||
|
||||
assertEquals(Authz.Action.SEND, FleetMcp.toolAction("fleet_send", Map.of()));
|
||||
assertEquals(Authz.Action.REPLY, FleetMcp.toolAction("fleet_reply", Map.of()));
|
||||
assertEquals(Authz.Action.ASK, FleetMcp.toolAction("fleet_ask", Map.of()));
|
||||
assertEquals(Authz.Action.READ, FleetMcp.toolAction("fleet_status", Map.of()));
|
||||
assertEquals(Authz.Action.DRAIN, FleetMcp.toolAction("fleet_ack", Map.of()));
|
||||
assertEquals(Authz.Action.SPAWN, FleetMcp.toolAction("fleet_spawn", Map.of()));
|
||||
assertEquals(Authz.Action.READ, FleetMcp.toolAction("fleet_list", Map.of()));
|
||||
assertEquals(Authz.Action.STOP, FleetMcp.toolAction("fleet_stop", Map.of()));
|
||||
assertEquals(Authz.Action.READ, FleetMcp.toolAction("fleet_profiles", Map.of()));
|
||||
assertEquals(Authz.Action.READ, FleetMcp.toolAction("fleet_whoami", Map.of()));
|
||||
assertEquals(Authz.Action.READ, FleetMcp.toolAction("fleet_poll", Map.of("ticket", "task")));
|
||||
assertEquals(Authz.Action.DRAIN, FleetMcp.toolAction("fleet_poll", Map.of("target", "term_b")));
|
||||
}
|
||||
|
||||
private static Set<String> toolsTheServerRegisters() {
|
||||
try {
|
||||
Matcher matcher = TOOL_REGISTRATION.matcher(Files.readString(MCP_SOURCE));
|
||||
Set<String> tools = new LinkedHashSet<>();
|
||||
while (matcher.find()) {
|
||||
tools.add(matcher.group(1));
|
||||
}
|
||||
return tools;
|
||||
} catch (Exception e) {
|
||||
throw new AssertionError("could not scrape FleetMcp tool registrations", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void aWorkerMayNotDrainAnotherSessionsInboxByPolling() {
|
||||
FleetMcp m = mcp(true);
|
||||
|
||||
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.EnumSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@@ -604,6 +605,60 @@ class FleetMcpTest {
|
||||
assertTrue(out.contains("\"reclaimable\":0"), out);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #284: {@code reclaimable} has exactly one definition, and this pins it over EVERY
|
||||
* {@link MemberSession.State} — so a state added later cannot slip through unconsidered. Both
|
||||
* views in a {@code fleet_list} response call {@link FleetMcp#reclaimable}, so they cannot
|
||||
* drift apart.
|
||||
*
|
||||
* <p>{@code BACKEND_ERROR} and {@code FAILED} are NOT reclaimable on purpose. The ticket asked
|
||||
* for them to be; that half of the ticket was wrong. Their seat is already out of
|
||||
* {@code live}, so it is already in {@code free} — counting it here too would report the same
|
||||
* seat twice.
|
||||
*/
|
||||
@Test
|
||||
void onlyReadyAndDoneSessionsAreReclaimable() {
|
||||
Set<MemberSession.State> expected = EnumSet.of(MemberSession.State.READY, MemberSession.State.DONE);
|
||||
for (MemberSession.State state : MemberSession.State.values()) {
|
||||
MemberSession session = new MemberSession("p1", "term1", "ltms-local", null,
|
||||
"/tmp", null, 0L, 0L, 0, state, null, null);
|
||||
assertEquals(expected.contains(state), FleetMcp.reclaimable(session, null),
|
||||
"state " + state + " must " + (expected.contains(state) ? "" : "not ")
|
||||
+ "count as reclaimable");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #284, the operator-visible half. {@code liveCount} here is the value the real counter
|
||||
* ({@code Fleetd.liveSessionCount}, proven against the actual spawn gate in
|
||||
* {@code FleetdBackendErrorSinkTest}) produces for this roster: 0, because both sessions are
|
||||
* terminal. What this test pins is what {@code fleet_list} says around it — the two seats show
|
||||
* up once, in {@code free}, and are NOT counted a second time as {@code reclaimable}; neither
|
||||
* is any member row; and both dead sessions are still listed so the lead can see why.
|
||||
*/
|
||||
@Test
|
||||
void terminalFailureSessionsFreeTheirSeatWithoutBeingCountedReclaimable() {
|
||||
FakeHerdr h = new FakeHerdr();
|
||||
SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
|
||||
MemberSession backendError = sessions.acquire("ltms-local", null, null, null);
|
||||
MemberSession failed = sessions.acquire("ltms-local", null, null, null);
|
||||
assertTrue(sessions.onBackendError(backendError.terminalId(), "backend exited"));
|
||||
sessions.onTurnFailed(failed.terminalId());
|
||||
|
||||
String out = textOf(FleetMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
|
||||
sessions, null, new FleetMcp.CapacitySource(profile -> 0, profile -> 2,
|
||||
() -> Set.of("ltms-local"), () -> 0), new FleetMcp.HealthCoverageSource(() -> "off"),
|
||||
FleetMcp.QuarantineSource.none(), Map.of(), ""));
|
||||
|
||||
assertTrue(out.contains("\"free\":2"), "both seats are back: " + out);
|
||||
assertTrue(out.contains("\"reclaimable\":0"),
|
||||
"the freed seats must not be counted a second time as reclaimable: " + out);
|
||||
assertFalse(out.contains("\"reclaimable\":true"),
|
||||
"no member row may claim a seat the profile count says is not held: " + out);
|
||||
assertTrue(out.contains("\"state\":\"backend_error\""), "the dead session stays visible: " + out);
|
||||
assertTrue(out.contains("\"state\":\"failed\""), "the failed session stays visible: " + out);
|
||||
}
|
||||
|
||||
@Test
|
||||
void inertCapacitySourceOmitsCapacityBlock() {
|
||||
FakeHerdr h = new FakeHerdr();
|
||||
|
||||
@@ -925,6 +925,71 @@ class ClaudeCodeLauncherTest {
|
||||
assertTrue(herdr.called("pane.close"), "stop via handle.id() must close the pane");
|
||||
}
|
||||
|
||||
/** A tab-placement launcher with {@code memberCredentials policy=allow-list} under a zsh shell — the
|
||||
* combination that makes {@link HerdrPeerLauncher#spawn} generate a real ZDOTDIR, so {@code
|
||||
* releaseZdotdir}'s effect (the directory's deletion) is observable from a test. */
|
||||
private ClaudeCodeLauncher serviceWithAllowList(FakeHerdr herdr) {
|
||||
FleetConfig.Profile cfg = new FleetConfig.Profile(
|
||||
"ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
|
||||
List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
|
||||
"worker: {profile} #{n}", null, null, null);
|
||||
Supplier<FleetConfig.MemberCredentials> creds = () -> new FleetConfig.MemberCredentials(
|
||||
FleetConfig.MemberCredentials.POLICY_ALLOW_LIST, List.of(), List.of(), null);
|
||||
Function<String, String> env = name -> "SHELL".equals(name) ? "/bin/zsh" : null;
|
||||
return new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(),
|
||||
env, 0, System::currentTimeMillis, () -> { }, null, creds);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #293: {@code stop()} used to run {@code spaces.closeTab} bare — any non-{@code
|
||||
* *_not_found} herdr error propagated straight out of {@code stop()}, skipping {@code
|
||||
* releaseZdotdir} entirely (the pane was already closed by that point, so the tab-close failure
|
||||
* is cosmetic, not a real teardown failure). Proves both halves of the fix: {@code stop()} no
|
||||
* longer throws for this failure, and {@code releaseZdotdir} still runs — observed here by the
|
||||
* generated ZDOTDIR actually being deleted, since {@code releaseZdotdir}'s last line is {@code
|
||||
* EnvAllowListScrub.deleteRecursively(dir)}.
|
||||
*/
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void stopStillReleasesZdotdirWhenCloseTabFailsWithANonNotFoundCode() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
ClaudeCodeLauncher svc = serviceWithAllowList(herdr);
|
||||
PeerHandle handle = svc.spawn(new SpawnRequest(null, null, null));
|
||||
Map<String, Object> tabCreateParams = (Map<String, Object>) herdr.lastCall("tab.create").params();
|
||||
Map<String, String> tabEnv = (Map<String, String>) tabCreateParams.get("env");
|
||||
String zdotdir = tabEnv.get("ZDOTDIR");
|
||||
assertNotNull(zdotdir, "policy=allow-list under a zsh shell must have generated a ZDOTDIR: " + tabEnv);
|
||||
Path dir = Path.of(zdotdir);
|
||||
assertTrue(Files.isDirectory(dir), "the generated ZDOTDIR must exist before stop(): " + dir);
|
||||
herdr.tabCloseFailsForTab("w9:t2", "internal_error");
|
||||
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
try {
|
||||
assertDoesNotThrow(() -> svc.stop(handle.id()),
|
||||
"fleetd #293: a failing tab.close is cosmetic — it must not propagate out of stop()");
|
||||
} finally {
|
||||
logger.detachAppender(appender);
|
||||
}
|
||||
|
||||
assertTrue(herdr.called("tab.close"), "tab.close was still attempted");
|
||||
assertFalse(Files.exists(dir),
|
||||
"releaseZdotdir must still run and delete the generated ZDOTDIR despite the tab.close "
|
||||
+ "failure: " + dir);
|
||||
String warn = appender.list.stream()
|
||||
.filter(e -> e.getLevel().equals(Level.WARN))
|
||||
.map(ILoggingEvent::getFormattedMessage)
|
||||
.filter(m -> m.contains("tab.close") && m.contains("w9:t2"))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
assertNotNull(warn, "the failing tab.close must be logged at WARN naming the tab id — a "
|
||||
+ "silently swallowed failure with no message is not an improvement. Log lines: "
|
||||
+ appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList());
|
||||
}
|
||||
|
||||
// --- CB-519: host-unique id, decoupled from the pane coordinate ------------------------------
|
||||
|
||||
@Test
|
||||
@@ -1100,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");
|
||||
@@ -1117,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 --------------------------------------
|
||||
@@ -2742,4 +2827,126 @@ class ClaudeCodeLauncherTest {
|
||||
"a WARN naming the cwd must fire when the profile sets no configDir: "
|
||||
+ appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList());
|
||||
}
|
||||
|
||||
// --- fleetd #285: seedTrustDialog under memberHerdrSocket --------------------------------------
|
||||
//
|
||||
// seedTrustDialog gated only on isProvisionedWorktree(cwd) and, being static, could not see
|
||||
// memberHerdrSocketConfigured() at all — unlike its sibling writeCharterFile one method below,
|
||||
// which already refuses the spawn when it cannot place the charter where a different-uid member
|
||||
// can read it. Under memberHerdrSocket + configDir unset, seedTrustDialog wrote fleetd's OWN
|
||||
// ~/.claude.json (the operator's real file) while believing it was seeding the member's. The fix
|
||||
// makes the method an instance method so it can see memberHerdrSocketConfigured(), and applies
|
||||
// the same "refuse, don't silently write somewhere wrong" rule writeCharterFile already uses.
|
||||
|
||||
@Test
|
||||
void seedTrustDialogUnderMemberHerdrSocketSharesTheFileWithTheConfiguredGroup(
|
||||
@TempDir Path configDir, @TempDir Path worktree, @TempDir Path worktreeRoot) throws Exception {
|
||||
markAsProvisionedWorktree(worktree);
|
||||
String group = currentUserGroup();
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
// trustProfile always sets a bridge mcpUrl, so a reply charter is generated too, which
|
||||
// means writeCharterFile ALSO runs under memberHerdrSocket and needs its own worktreeRoot —
|
||||
// pass one so this test isolates the trust-seed behaviour instead of tripping that refusal.
|
||||
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
|
||||
serviceWithConfig(herdr, cfg, () -> configWithMemberHerdrSocket(worktreeRoot.toString(), group)).spawn();
|
||||
|
||||
Path claudeJson = configDir.resolve(".claude.json");
|
||||
assertTrue(Files.exists(claudeJson), "still seeded into <configDir>/.claude.json under memberHerdrSocket");
|
||||
JsonNode project = new ObjectMapper().readTree(claudeJson.toFile())
|
||||
.path("projects").path(worktree.toString());
|
||||
assertTrue(project.path("hasTrustDialogAccepted").asBoolean(false));
|
||||
|
||||
assertEquals("rw-r-----", PosixFilePermissions.toString(Files.getPosixFilePermissions(claudeJson)),
|
||||
"under memberHerdrSocket the file must be shared group-readable, mirroring the "
|
||||
+ "charter file's own per-file mode (fleetd #222) — a 0600 file (Claude "
|
||||
+ "Code's own default) is unreadable by the member's different OS user");
|
||||
String actualGroup = Files.getFileAttributeView(claudeJson, PosixFileAttributeView.class)
|
||||
.readAttributes().group().getName();
|
||||
assertEquals(group, actualGroup, "the file must be chgrp'd to the configured worktreeGroup");
|
||||
}
|
||||
|
||||
/**
|
||||
* The bug itself: {@code memberHerdrSocket} configured, {@code configDir} unset. Before the fix
|
||||
* this wrote fleetd's own default {@code ~/.claude.json} (here redirected to {@code fakeHome} so
|
||||
* a reintroduced bug still cannot touch the real operator file); after the fix it must refuse the
|
||||
* spawn instead, naming {@code configDir} as the missing key, before the member is ever started.
|
||||
*/
|
||||
@Test
|
||||
void seedTrustDialogUnderMemberHerdrSocketRefusesWhenConfigDirUnset(
|
||||
@TempDir Path fakeHome, @TempDir Path worktree) throws Exception {
|
||||
markAsProvisionedWorktree(worktree);
|
||||
String originalHome = System.getProperty("user.home");
|
||||
System.setProperty("user.home", fakeHome.toString());
|
||||
try {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FleetConfig.Profile cfg = trustProfile(null, worktree.toString());
|
||||
ClaudeCodeLauncher launcher = serviceWithConfig(herdr, cfg,
|
||||
() -> configWithMemberHerdrSocket(null, "some-group"));
|
||||
|
||||
IllegalStateException ex = assertThrows(IllegalStateException.class, launcher::spawn,
|
||||
"memberHerdrSocket + no configDir must refuse the spawn, not write fleetd's own "
|
||||
+ "default ~/.claude.json");
|
||||
assertTrue(ex.getMessage().contains("configDir"),
|
||||
"the refusal must name the missing config key — got: " + ex.getMessage());
|
||||
assertFalse(Files.exists(fakeHome.resolve(".claude.json")),
|
||||
"nothing may be written to fleetd's own default home — this is the exact fleetd "
|
||||
+ "#285 defect: writing the operator's own home instead of the member's");
|
||||
assertFalse(herdr.called("agent.start"),
|
||||
"the spawn must be refused BEFORE the member is ever started — got calls: " + herdr.calls);
|
||||
} finally {
|
||||
System.setProperty("user.home", originalHome);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@code configDir} alone is not enough — without {@code worktreeGroup} the file fleetd writes
|
||||
* stays {@code 0600} and the member's different OS user still cannot read it, so this must also
|
||||
* refuse, naming {@code worktreeGroup} this time.
|
||||
*/
|
||||
@Test
|
||||
void seedTrustDialogUnderMemberHerdrSocketRefusesWhenWorktreeGroupUnset(
|
||||
@TempDir Path configDir, @TempDir Path worktree) throws Exception {
|
||||
markAsProvisionedWorktree(worktree);
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
|
||||
ClaudeCodeLauncher launcher = serviceWithConfig(herdr, cfg,
|
||||
() -> configWithMemberHerdrSocket(null, null));
|
||||
|
||||
IllegalStateException ex = assertThrows(IllegalStateException.class, launcher::spawn,
|
||||
"memberHerdrSocket + no worktreeGroup must refuse the spawn, not write an unreadable file");
|
||||
assertTrue(ex.getMessage().contains("worktreeGroup"),
|
||||
"configDir alone is not enough — got: " + ex.getMessage());
|
||||
assertFalse(Files.exists(configDir.resolve(".claude.json")),
|
||||
"nothing may be written when the file cannot be shared with the member's group");
|
||||
assertFalse(herdr.called("agent.start"),
|
||||
"the spawn must be refused BEFORE the member is ever started");
|
||||
}
|
||||
|
||||
/**
|
||||
* Regression proof: with {@code memberHerdrSocket} ABSENT — even given a LIVE, non-null {@code
|
||||
* config} supplier (not merely {@code config == null}, which every other seedTrustDialog test in
|
||||
* this file already exercises) — the write must stay byte-identical to before this fix: existing
|
||||
* {@code 0600} permissions preserved, no chgrp/chmod attempted. This is the proof the ticket asks
|
||||
* for: the path the whole live fleet uses today is unchanged by this fix.
|
||||
*/
|
||||
@Test
|
||||
void seedTrustDialogPreservesExisting0600PermissionsWhenMemberHerdrSocketAbsentEvenWithALiveConfigSupplier(
|
||||
@TempDir Path configDir, @TempDir Path worktree) throws Exception {
|
||||
markAsProvisionedWorktree(worktree);
|
||||
Path claudeJson = configDir.resolve(".claude.json");
|
||||
Files.writeString(claudeJson, "{}");
|
||||
assumeTrue(Files.getFileAttributeView(claudeJson, PosixFileAttributeView.class) != null,
|
||||
"no POSIX permissions on this filesystem — skipping rather than failing");
|
||||
Files.setPosixFilePermissions(claudeJson, PosixFilePermissions.fromString("rw-------"));
|
||||
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
|
||||
FleetConfig config = new FleetConfig(null, null, null, Map.of(), null, null, null, null, null,
|
||||
null, null, null, null, null, null, null, null, null, null, null, null, null).withDefaults();
|
||||
serviceWithConfig(herdr, cfg, () -> config).spawn();
|
||||
|
||||
assertEquals("rw-------", PosixFilePermissions.toString(Files.getPosixFilePermissions(claudeJson)),
|
||||
"with memberHerdrSocket absent — even given a live config supplier — the seed must "
|
||||
+ "stay byte-identical to before this fix: no chgrp/chmod attempted");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -975,6 +975,64 @@ class MessageServiceTest {
|
||||
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #282: a worker that chains a SECOND {@code fleet_ask} inside the same resumed turn —
|
||||
* before it ever calls {@code fleet_reply} — used to kill its own async ticket. {@code answer()}
|
||||
* opens a fresh forward waiter but (unlike {@code send()}) never registered it in
|
||||
* {@code asyncTasksByWaiter}, so the second ask's {@code markAsyncQuestion} found no {@code Task}
|
||||
* to re-associate. That waiter still resolved with the second {@code QUESTION} once the worker
|
||||
* asked again, and {@code answer()} completed the ticket's future with that QUESTION "reply"
|
||||
* unconditionally — so {@code fleet_poll} reported FAILED while the worker was still alive and
|
||||
* the primary was mid-conversation with it.
|
||||
*
|
||||
* <p>Driven entirely through {@code MessageService}'s public API (sendAsync/ask/answer/poll) —
|
||||
* never by reaching into {@link Rendezvous} or the task maps directly, so this test cannot pass
|
||||
* for a reason unrelated to the real bug.
|
||||
*/
|
||||
@Test
|
||||
void secondFleetAskInTheSameResumedTurnDoesNotKillTheAsyncTicket() throws Exception {
|
||||
String ticket = messages.sendAsync(T, "task that asks twice");
|
||||
awaitWaiting();
|
||||
|
||||
// The worker's first fleet_ask.
|
||||
CompletableFuture<MessageService.AskResult> ask1 =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "Q1", 5000));
|
||||
MessageService.TaskView asking1 = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
|
||||
assertEquals("Q1", asking1.reply());
|
||||
|
||||
// The primary answers it — answer() resumes the turn and blocks for what comes next.
|
||||
CompletableFuture<MessageService.Reply> answer1 = CompletableFuture.supplyAsync(
|
||||
() -> messages.answer(asking1.turnId(), "a1", 5000));
|
||||
assertEquals("a1", ask1.get(5, TimeUnit.SECONDS).answer());
|
||||
|
||||
// Still in the SAME resumed turn — before replying — the worker asks again.
|
||||
CompletableFuture<MessageService.AskResult> ask2 =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "Q2", 5000));
|
||||
|
||||
// answer1's own call unblocks with the second QUESTION (documented QUESTION-chaining
|
||||
// behaviour — see FleetMcp.answer's javadoc: "Answer it by calling fleet_send again with
|
||||
// turnId=..."). The bug: this used to also kill the async ticket in the process.
|
||||
MessageService.Reply firstAnswerResult = answer1.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.QUESTION, firstAnswerResult.outcome());
|
||||
String turnId2 = firstAnswerResult.turnId();
|
||||
|
||||
MessageService.TaskView asking2 = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
|
||||
assertEquals("Q2", asking2.reply(),
|
||||
"the ticket must surface the SECOND question, not be dead/FAILED");
|
||||
assertEquals(turnId2, asking2.turnId());
|
||||
|
||||
// The primary answers the second question; the worker finally sends its real fleet_reply.
|
||||
CompletableFuture<MessageService.Reply> answer2 = CompletableFuture.supplyAsync(
|
||||
() -> messages.answer(turnId2, "a2", 5000));
|
||||
assertEquals("a2", ask2.get(5, TimeUnit.SECONDS).answer());
|
||||
awaitWaiting();
|
||||
assertTrue(rendezvous.resolve(T, "done"));
|
||||
assertEquals(MessageService.Outcome.REPLIED, answer2.get(5, TimeUnit.SECONDS).outcome());
|
||||
|
||||
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
|
||||
assertEquals("done", done.reply());
|
||||
}
|
||||
|
||||
// --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
|
||||
|
||||
@Test
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package dev.ltms.fleet.rest;
|
||||
|
||||
import dev.ltms.fleet.auth.CallerResolver;
|
||||
import dev.ltms.fleet.auth.Authz;
|
||||
import dev.ltms.fleet.auth.MemberRegistry;
|
||||
import dev.ltms.fleet.config.FleetConfig;
|
||||
import dev.ltms.fleet.guard.SubscriptionGuard;
|
||||
@@ -25,8 +26,15 @@ import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
@@ -36,6 +44,10 @@ import static org.junit.jupiter.api.Assertions.*;
|
||||
*/
|
||||
class FleetAppAuthTest {
|
||||
|
||||
private static final Path REST_SOURCE = Path.of("src/main/java/dev/ltms/fleet/rest/FleetApp.java");
|
||||
private static final Pattern ROUTE_REGISTRATION =
|
||||
Pattern.compile("app\\.(get|post|delete|put|patch)\\(\\s*\"([^\"]+)\"");
|
||||
|
||||
private final HttpClient http = HttpClient.newHttpClient();
|
||||
private Javalin app;
|
||||
private Metrics metrics;
|
||||
@@ -92,6 +104,49 @@ class FleetAppAuthTest {
|
||||
return http.send(b.build(), HttpResponse.BodyHandlers.ofString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void everyRegisteredRouteHasItsHandlerActionPinned() {
|
||||
Set<String> registered = routesTheServerRegisters();
|
||||
assertTrue(registered.size() >= 15,
|
||||
"scraped only " + registered.size() + " route registrations from FleetApp (" + registered
|
||||
+ "); the app.<verb>(\"…\") scrape has stopped matching");
|
||||
// Liveness must work before credentials can be checked, so this route is deliberately open.
|
||||
assertTrue(registered.remove("GET /healthz"), "GET /healthz must stay an explicit ungated exception");
|
||||
registered.forEach(route -> assertDoesNotThrow(() -> FleetApp.routeAction(route),
|
||||
() -> route + " is registered but has no pinned authorization action"));
|
||||
assertEquals(Authz.Action.METRICS, FleetApp.routeAction("GET /metrics"));
|
||||
assertEquals(Authz.Action.SPAWN, FleetApp.routeAction("POST /members"));
|
||||
assertEquals(Authz.Action.STOP, FleetApp.routeAction("DELETE /members/{paneId}"));
|
||||
assertEquals(Authz.Action.SEND, FleetApp.routeAction("POST /sessions/{id}/message"));
|
||||
assertEquals(Authz.Action.REPLY, FleetApp.routeAction("POST /sessions/{id}/reply"));
|
||||
assertEquals(Authz.Action.DRAIN, FleetApp.routeAction("GET /sessions/{id}/replies"));
|
||||
assertEquals(Authz.Action.ASK, FleetApp.routeAction("POST /sessions/{id}/ask"));
|
||||
for (String route : Set.of("GET /sessions", "GET /agents", "GET /members", "GET /profiles",
|
||||
"GET /member-credentials", "GET /sessions/{id}/status", "GET /tasks/{ticket}")) {
|
||||
assertEquals(Authz.Action.READ, FleetApp.routeAction(route), route);
|
||||
}
|
||||
assertThrows(IllegalArgumentException.class, () -> FleetApp.routeAction("GET /healthz"));
|
||||
}
|
||||
|
||||
private static Set<String> routesTheServerRegisters() {
|
||||
try {
|
||||
String source = Files.readString(REST_SOURCE).lines()
|
||||
.filter(line -> {
|
||||
String stripped = line.stripLeading();
|
||||
return !(stripped.startsWith("//") || stripped.startsWith("*") || stripped.startsWith("/*"));
|
||||
})
|
||||
.collect(Collectors.joining("\n"));
|
||||
Matcher matcher = ROUTE_REGISTRATION.matcher(source);
|
||||
Set<String> routes = new LinkedHashSet<>();
|
||||
while (matcher.find()) {
|
||||
routes.add(matcher.group(1).toUpperCase(Locale.ROOT) + " " + matcher.group(2));
|
||||
}
|
||||
return routes;
|
||||
} catch (Exception e) {
|
||||
throw new AssertionError("could not scrape FleetApp route registrations", e);
|
||||
}
|
||||
}
|
||||
|
||||
// --- loopback-trust: the caller is the primary -------------------------------------------
|
||||
|
||||
@Test
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -17,6 +17,9 @@ public final class FakeWorktrees implements Worktrees {
|
||||
public record RemoveCall(String repoRoot, String worktreePath) {
|
||||
}
|
||||
|
||||
public record DeleteBranchCall(String repoRoot, String branch) {
|
||||
}
|
||||
|
||||
public record OverlayCall(String repoRoot, String worktreePath,
|
||||
List<String> requested, List<String> copied, List<String> skipWorktree) {
|
||||
}
|
||||
@@ -35,6 +38,7 @@ public final class FakeWorktrees implements Worktrees {
|
||||
|
||||
private final List<AddCall> addCalls = new CopyOnWriteArrayList<>();
|
||||
private final List<RemoveCall> removeCalls = new CopyOnWriteArrayList<>();
|
||||
private final List<DeleteBranchCall> deleteBranchCalls = new CopyOnWriteArrayList<>();
|
||||
private final List<OverlayCall> overlayCalls = new CopyOnWriteArrayList<>();
|
||||
private final List<RepoRootCall> repoRootCalls = new CopyOnWriteArrayList<>();
|
||||
private final List<SnapshotCall> snapshotCalls = new CopyOnWriteArrayList<>();
|
||||
@@ -51,6 +55,8 @@ public final class FakeWorktrees implements Worktrees {
|
||||
private final AtomicLong snapshotSeq = new AtomicLong();
|
||||
private volatile RuntimeException addFailure;
|
||||
private volatile RuntimeException snapshotFailure;
|
||||
private volatile RuntimeException removeFailure;
|
||||
private volatile RuntimeException overlayFailure;
|
||||
private volatile boolean dirty = false;
|
||||
private volatile String repoRoot = "/repo";
|
||||
private volatile String prefix = "/worktrees";
|
||||
@@ -98,6 +104,21 @@ public final class FakeWorktrees implements Worktrees {
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make subsequent {@link #remove} calls throw (fleetd #283: a stale index lock, a slow
|
||||
* filesystem, or {@code remove}'s own 30s exec timeout escaping the last, previously bare,
|
||||
* step of {@link SessionManager#release}). */
|
||||
public FakeWorktrees failRemove(String message) {
|
||||
this.removeFailure = new WorktreeException(message);
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make subsequent {@link #overlayParity} calls throw (simulates a post-{@code add()} spawn
|
||||
* failure — fleetd #283 defect 2 — so the {@code acquireWithWorktree} catch runs). */
|
||||
public FakeWorktrees failOverlay(String message) {
|
||||
this.overlayFailure = new WorktreeException(message);
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Configure the value returned by {@link #wipRefs}. */
|
||||
public FakeWorktrees withWipRefs(WipRefStats stats) {
|
||||
this.wipRefs = stats;
|
||||
@@ -132,6 +153,14 @@ public final class FakeWorktrees implements Worktrees {
|
||||
@Override
|
||||
public void remove(String repoRoot, String worktreePath) {
|
||||
removeCalls.add(new RemoveCall(repoRoot, worktreePath));
|
||||
if (removeFailure != null) {
|
||||
throw removeFailure;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteBranch(String repoRoot, String branch) {
|
||||
deleteBranchCalls.add(new DeleteBranchCall(repoRoot, branch));
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -146,6 +175,9 @@ public final class FakeWorktrees implements Worktrees {
|
||||
|
||||
@Override
|
||||
public void overlayParity(String repoRoot, String worktreePath, List<String> overlay) {
|
||||
if (overlayFailure != null) {
|
||||
throw overlayFailure;
|
||||
}
|
||||
List<String> copied = new java.util.ArrayList<>();
|
||||
List<String> skipped = new java.util.ArrayList<>();
|
||||
for (String rel : overlay) {
|
||||
@@ -218,6 +250,14 @@ public final class FakeWorktrees implements Worktrees {
|
||||
return removeCalls.isEmpty() ? null : removeCalls.getLast();
|
||||
}
|
||||
|
||||
public List<DeleteBranchCall> deleteBranchCalls() {
|
||||
return List.copyOf(deleteBranchCalls);
|
||||
}
|
||||
|
||||
public DeleteBranchCall lastDeleteBranch() {
|
||||
return deleteBranchCalls.isEmpty() ? null : deleteBranchCalls.getLast();
|
||||
}
|
||||
|
||||
public OverlayCall lastOverlay() {
|
||||
return overlayCalls.isEmpty() ? null : overlayCalls.getLast();
|
||||
}
|
||||
|
||||
@@ -74,6 +74,7 @@ class SessionManagerTest {
|
||||
*/
|
||||
private static final class RecordingWorktrees implements Worktrees {
|
||||
private final List<String> removeCalls = new java.util.ArrayList<>();
|
||||
private final List<String> deleteBranchCalls = new java.util.ArrayList<>();
|
||||
private final List<String> snapshotCalls = new java.util.ArrayList<>();
|
||||
private final java.util.Set<String> failRemoveFor = new java.util.HashSet<>();
|
||||
private volatile boolean dirty = false;
|
||||
@@ -114,6 +115,11 @@ class SessionManagerTest {
|
||||
removeCalls.add(worktreePath);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteBranch(String repoRoot, String branch) {
|
||||
deleteBranchCalls.add(branch);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean hasUncommitted(String worktreePath) {
|
||||
if (hasUncommittedFailure != null) {
|
||||
@@ -158,6 +164,10 @@ class SessionManagerTest {
|
||||
return List.copyOf(removeCalls);
|
||||
}
|
||||
|
||||
List<String> deleteBranchCalls() {
|
||||
return List.copyOf(deleteBranchCalls);
|
||||
}
|
||||
|
||||
List<String> snapshotCalls() {
|
||||
return List.copyOf(snapshotCalls);
|
||||
}
|
||||
@@ -989,8 +999,21 @@ class SessionManagerTest {
|
||||
+ "dirty check threw");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #283 defect 1 changed this test's own premise, so its assertions are updated along
|
||||
* with the production fix. Before #283, the middle session's worktree-removal failure escaped
|
||||
* {@code release()} uncaught, and this test proved {@code reapIdle}'s own per-session try/catch
|
||||
* (CB-581) kept the rest of the pass going regardless. Now that {@code release()} itself catches
|
||||
* a worktree-removal failure (matching every sibling cleanup step in that method) and only logs
|
||||
* a WARN, {@code release()} no longer throws for this reason — so all three idle sessions are
|
||||
* released and counted, and the middle one's removal failure is now visible only as the WARN
|
||||
* {@code release()} itself logs, not as a reap-loop catch. {@code reapIdle}'s own guard (for a
|
||||
* failure {@code release()} still cannot swallow, e.g. from {@code launcher.stop}) is untouched
|
||||
* by this ticket. This test no longer exercises that guard — reaching it now needs a failure
|
||||
* that #283 does not catch inside {@code release()} itself.
|
||||
*/
|
||||
@Test
|
||||
void reapIdleSurvivesOneSessionThatFailsToRelease() {
|
||||
void reapIdleCountsAllThreeSessionsWhenOnlyItsWorktreeRemovalFails() {
|
||||
long[] clock = {0};
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
RecordingWorktrees worktrees = new RecordingWorktrees();
|
||||
@@ -1004,8 +1027,8 @@ class SessionManagerTest {
|
||||
sessions.asPresence().markPresent(a.terminalId());
|
||||
sessions.asPresence().markPresent(b.terminalId());
|
||||
sessions.asPresence().markPresent(c.terminalId());
|
||||
// The middle session's worktree removal fails — release() propagates that, so this is the
|
||||
// one call reapIdle's per-session guard must survive without skipping the rest of the pass.
|
||||
// The middle session's worktree removal fails — fleetd #283 makes release() catch and log
|
||||
// this itself, so it no longer propagates out of release() at all.
|
||||
worktrees.failRemoveFor(b.worktree());
|
||||
|
||||
LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
|
||||
@@ -1026,14 +1049,16 @@ class SessionManagerTest {
|
||||
.map(ILoggingEvent::getFormattedMessage)
|
||||
.filter(m -> m.contains(b.paneId()))
|
||||
.findFirst()
|
||||
.orElse("no reap-failure WARN logged");
|
||||
.orElse("no worktree-removal-failure WARN logged");
|
||||
assertTrue(warn.contains(b.terminalId()), "the WARN names the failed session's terminal: " + warn);
|
||||
assertTrue(warn.contains(b.worktree()), "the WARN names the failed session's worktree: " + warn);
|
||||
} finally {
|
||||
sessionLog.detachAppender(appender);
|
||||
}
|
||||
|
||||
assertEquals(2, reaped, "the middle session's failure is logged, not counted as reaped");
|
||||
assertEquals(3, reaped,
|
||||
"fleetd #283: release() no longer throws for a worktree-removal failure, so reapIdle "
|
||||
+ "counts all three idle sessions as reaped");
|
||||
assertTrue(sessions.get(a.paneId()).isEmpty(), "the first session is still released");
|
||||
assertTrue(sessions.get(c.paneId()).isEmpty(), "the third session is still released");
|
||||
assertTrue(sessions.get(b.paneId()).isEmpty(),
|
||||
@@ -1044,6 +1069,79 @@ class SessionManagerTest {
|
||||
assertEquals(1, paneCloseCallsFor(herdr, "w9:pRoot_3"), "the third pane is stopped");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #290: the #283 fix above closed the one trigger this suite used for {@code
|
||||
* reapIdle}'s own per-session try/catch (CB-581) — a worktree-removal failure is now caught
|
||||
* and logged inside {@code release()} itself, so it never reaches {@code reapIdle}'s guard at
|
||||
* all. This test restores coverage of that guard using the trigger the ticket names: {@code
|
||||
* release()} calls {@code launcher.stop(paneId)} with no try/catch around it, so a failing
|
||||
* {@code pane.close} propagates straight out of {@code release()} uncaught. {@link
|
||||
* FakeHerdr#paneCloseFailsForPane} (added for this ticket) makes exactly the middle session's
|
||||
* stop fail, while the other two still succeed, so this proves {@code reapIdle} keeps reaping
|
||||
* the rest of the roster rather than aborting the whole pass.
|
||||
*/
|
||||
@Test
|
||||
void reapIdleSurvivesOneSessionWhoseLauncherStopFails() {
|
||||
long[] clock = {0};
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
RecordingWorktrees worktrees = new RecordingWorktrees();
|
||||
SessionManager sessions = sessionManager(herdr, worktrees, () -> clock[0]);
|
||||
MemberSession a = sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-290a", null));
|
||||
MemberSession b = sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-290b", null));
|
||||
MemberSession c = sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-290c", null));
|
||||
sessions.asPresence().markPresent(a.terminalId());
|
||||
sessions.asPresence().markPresent(b.terminalId());
|
||||
sessions.asPresence().markPresent(c.terminalId());
|
||||
// Only the middle session's herdr pane fails to close — a and c stop normally. This is the
|
||||
// trigger reapIdle's own guard is for, now that #283 closed the worktree-removal trigger.
|
||||
herdr.paneCloseFailsForPane("w9:pRoot_2", "internal_error");
|
||||
|
||||
LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
|
||||
ch.qos.logback.classic.Logger sessionLog =
|
||||
(ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.setContext(ctx);
|
||||
appender.start();
|
||||
sessionLog.addAppender(appender);
|
||||
sessionLog.setLevel(Level.WARN);
|
||||
int reaped;
|
||||
try {
|
||||
clock[0] = 100;
|
||||
reaped = sessions.reapIdle(10);
|
||||
|
||||
String warn = appender.list.stream()
|
||||
.filter(e -> e.getLevel().equals(Level.WARN))
|
||||
.map(ILoggingEvent::getFormattedMessage)
|
||||
.filter(m -> m.contains("reap failed") && m.contains(b.paneId()))
|
||||
.findFirst()
|
||||
.orElse("no reap-failed WARN logged for the failing session");
|
||||
assertTrue(warn.contains(b.terminalId()), "the WARN names the failed session's terminal: " + warn);
|
||||
} finally {
|
||||
sessionLog.detachAppender(appender);
|
||||
}
|
||||
|
||||
assertEquals(2, reaped,
|
||||
"the middle session's launcher.stop failure is not counted as reaped, but must not "
|
||||
+ "abort reaping the other two");
|
||||
assertTrue(sessions.get(a.paneId()).isEmpty(), "the first session is still released");
|
||||
assertTrue(sessions.get(c.paneId()).isEmpty(),
|
||||
"the third session is still reached and released — proves the pass did not abort "
|
||||
+ "when the middle session's release() threw");
|
||||
assertTrue(sessions.get(b.paneId()).isEmpty(),
|
||||
"the middle session is still deregistered — release() removes it from the registry "
|
||||
+ "before launcher.stop() runs, regardless of whether stop() then throws");
|
||||
assertEquals(1, paneCloseCallsFor(herdr, "w9:pRoot_1"), "the first pane is stopped");
|
||||
assertEquals(1, paneCloseCallsFor(herdr, "w9:pRoot_2"),
|
||||
"the middle pane's stop was attempted, even though it failed");
|
||||
assertEquals(1, paneCloseCallsFor(herdr, "w9:pRoot_3"), "the third pane is stopped");
|
||||
assertEquals(List.of(a.worktree(), c.worktree()), worktrees.removeCalls().stream().sorted().toList(),
|
||||
"the middle session's worktree removal never runs — release() throws before reaching "
|
||||
+ "it — while the other two, unaffected, still have theirs removed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void unchangedRegressionCleanCompletedReleaseStillRemovesTheWorktree() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
@@ -1058,6 +1156,41 @@ class SessionManagerTest {
|
||||
"COMPLETED release of a clean worktree still removes it");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #293: {@code HerdrPeerLauncher.stop()} used to run {@code spaces.closeTab} bare — a
|
||||
* failing {@code tab.close} (any code other than {@code *_not_found}) propagated straight out
|
||||
* of {@code stop()}. {@code SessionManager.release} calls {@code launcher.stop(paneId)} with
|
||||
* no try/catch (fleetd #283 wrapped the WORKTREE-removal step further down, not this one), so
|
||||
* the throw happened <em>before</em> that worktree-removal step ever ran — and by then {@code
|
||||
* registry.remove(paneId)} had already run, so a second {@code stop} is a no-op: the worktree
|
||||
* leaked with no retry path. The pane itself is already closed by the time {@code tab.close}
|
||||
* runs, so its failure is cosmetic workspace tidying, not a real teardown failure. The fix
|
||||
* wraps {@code closeTab} inside {@code stop()} so it no longer throws for this reason; this
|
||||
* test proves both halves at once: {@code release()} does not throw, and it still removes the
|
||||
* worktree.
|
||||
*/
|
||||
@Test
|
||||
void releaseStillRemovesTheWorktreeWhenCloseTabFails() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
RecordingWorktrees worktrees = new RecordingWorktrees();
|
||||
SessionManager sessions = sessionManager(herdr, worktrees);
|
||||
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-293a", null));
|
||||
// FakeHerdr's pane.get always answers with tab_id "w9:t2" for a tab-placement spawn.
|
||||
herdr.tabCloseFailsForTab("w9:t2", "internal_error");
|
||||
|
||||
assertDoesNotThrow(() -> sessions.release(s.paneId()),
|
||||
"a failing tab.close is cosmetic (the pane is already closed by then) — it must not "
|
||||
+ "propagate out of release()");
|
||||
|
||||
assertTrue(herdr.called("tab.close"), "tab.close was still attempted");
|
||||
assertEquals(List.of(s.worktree()), worktrees.removeCalls(),
|
||||
"release() must still remove the worktree even though tab.close failed — this is "
|
||||
+ "the leak fleetd #293 reports: before the fix, release() never reached this "
|
||||
+ "step at all");
|
||||
assertTrue(sessions.get(s.paneId()).isEmpty(), "the session is still deregistered");
|
||||
}
|
||||
|
||||
@Test
|
||||
void unchangedRegressionDirtyCompletedReleaseStillPreservesTheWorktree() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
@@ -212,9 +212,41 @@ class WorktreeSessionManagerTest {
|
||||
assertEquals("/repo", remove.repoRoot());
|
||||
assertEquals(s.worktree(), remove.worktreePath());
|
||||
// The fake records no branch-delete calls because Worktrees.remove only removes the checkout.
|
||||
assertTrue(worktrees.deleteBranchCalls().isEmpty(),
|
||||
"a normal release must NEVER delete the branch — it is the worker's only recoverable "
|
||||
+ "copy of committed work, and only the failed-provisioning path may remove it");
|
||||
assertTrue(sessions.get(paneId).isEmpty(), "released session is no longer retrievable");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #283 defect 1. Every other cleanup step in {@code release()} is wrapped in try/catch,
|
||||
* because {@code exec()} can throw on a non-zero exit or its own 30s timeout — this was the one
|
||||
* step left bare. By the time it runs, the registry entry, the retained handle, and the pane are
|
||||
* all already gone, so a throw here used to escape {@code release()} after the session was
|
||||
* already fully torn down: a second stop on the same paneId is a no-op (nothing left to find),
|
||||
* so there was no retry path, and the caller saw a failed stop for a session that was in fact
|
||||
* gone. This test makes the worktree removal throw and asserts release() still completes with
|
||||
* the pane stopped and the registry clean.
|
||||
*/
|
||||
@Test
|
||||
void releaseCompletesAndStopsPaneEvenWhenWorktreeRemovalThrows() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
|
||||
.failRemove("stale index lock");
|
||||
SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
|
||||
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-283-1", null));
|
||||
String paneId = s.paneId();
|
||||
|
||||
sessions.release(paneId); // must not throw
|
||||
|
||||
assertTrue(herdr.called("pane.close"), "the pane is still stopped despite the removal failure");
|
||||
assertEquals(1, worktrees.removeCalls().size(), "worktree removal was still attempted");
|
||||
assertTrue(sessions.get(paneId).isEmpty(),
|
||||
"the session is deregistered regardless of the removal failure");
|
||||
assertEquals(0, sessions.size(), "the registry is left clean");
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-576. A normal {@code COMPLETED} release whose worktree holds uncommitted work must NOT
|
||||
* remove it — {@code --force} would destroy the worker's only copy. The bridge cannot see
|
||||
@@ -352,6 +384,39 @@ class WorktreeSessionManagerTest {
|
||||
assertEquals(0, sessions.size(), "failed acquire leaves no registry entry");
|
||||
assertFalse(herdr.called("agent.start"), "spawn is never reached when add fails");
|
||||
assertTrue(worktrees.removeCalls().isEmpty(), "no worktree was added, so none is removed");
|
||||
assertTrue(worktrees.deleteBranchCalls().isEmpty(),
|
||||
"add() itself never created the branch in git, so there is nothing to delete");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #283 defect 2. {@code acquireWithWorktree}'s catch covers every failure AFTER
|
||||
* {@code worktrees.add()} returns — {@code overlayParity}, {@code shareWithGroup},
|
||||
* {@code launcher.spawn} itself — so by the time it runs, {@code branch} was actually created in
|
||||
* git. It removed only the worktree and forgot the branch, leaking a {@code worker/<slug>-<nonce>}
|
||||
* branch on every routine spawn failure (a quarantined credential, a backend refusal). This test
|
||||
* makes {@code overlayParity} (a post-add() step) throw and asserts the branch is deleted, the
|
||||
* same way #274 already does for the sibling failure inside {@code add()} itself.
|
||||
*/
|
||||
@Test
|
||||
void spawnFailureAfterAddDeletesTheOrphanedBranch() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
|
||||
.failOverlay("overlayParity failed");
|
||||
SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
|
||||
|
||||
assertThrows(WorktreeException.class, () ->
|
||||
sessions.acquire("ltms-local", null, "/caller/proj", null,
|
||||
new WorktreeRequest("cb-283-2", null)));
|
||||
|
||||
assertEquals(0, sessions.size(), "failed acquire leaves no registry entry");
|
||||
assertFalse(herdr.called("agent.start"), "spawn is never reached when overlayParity fails");
|
||||
assertEquals(1, worktrees.removeCalls().size(), "the worktree checkout is still removed");
|
||||
assertEquals(1, worktrees.deleteBranchCalls().size(),
|
||||
"the orphaned branch that add() actually created must also be deleted");
|
||||
FakeWorktrees.DeleteBranchCall del = worktrees.lastDeleteBranch();
|
||||
assertEquals("/repo", del.repoRoot());
|
||||
FakeWorktrees.AddCall add = worktrees.lastAdd();
|
||||
assertEquals(add.branch(), del.branch(), "the branch deleted is the exact one add() created");
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
Reference in New Issue
Block a user