Compare commits

..

1 Commits

Author SHA1 Message Date
Dai Ha 2dff4b84a4 Audit: async ticket / rendezvous lifecycle in fleetd msg package 2026-09-04 10:35:01 +07:00
22 changed files with 157 additions and 965 deletions
+96
View File
@@ -0,0 +1,96 @@
# Audit: async ticket / rendezvous lifecycle (`fleetd/src/main/java/dev/ltms/fleet/msg/`)
Scope: `Rendezvous.java` and `MessageService.java` — the lifecycle of an async ticket
(`Task`) and a rendezvous waiter: create, send, ask, answer, resolve, timeout, abandon, prune.
## Main finding
```
1. fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java:938
2. issue: answer() completes an async ticket's future with a QUESTION outcome when the worker
asks a second fleet_ask in the same resumed turn, permanently mislabeling a live delegation
as failed and losing its real reply to fleet_poll.
3. fix: guard the finishAsyncTask(turnId, result) call at line 938 the same way sendAsync's
lambda already guards its own call (lines 999-1005): skip it when result.outcome() ==
Outcome.QUESTION, and instead re-associate the task with the new turnId (as markAsyncQuestion
does on the first ask).
4. severity: high
```
### Call sequence that reaches it
1. Lead: `fleet_send{sessionId: W, content: "task", wait:false}` → `sendAsync` creates `task1`
/ `ticket1`. Its worker thread calls `send(W, content, ASYNC_TIMEOUT_MS, onAccepted, task1)`,
which does `asyncTasksByWaiter.put(reply, task1)` (line 802) before blocking on
`reply.get()`.
2. Worker `W` calls `fleet_ask{"Q1"}` → `ask(W, "Q1", t)`. `markAsyncQuestion` finds `task1`
via `asyncTasksByWaiter`, stamps `task1.turnId = turnId1`,
`asyncTasksByTurn[turnId1] = task1`. `resolveQuestion` wakes step 1's `send()`, which returns
`Outcome.QUESTION`; `sendAsync`'s lambda sees `QUESTION` and deliberately does **not** call
`finishAsyncTask` (lines 1000-1005) — `ticket1` correctly polls `Phase.ASKING`.
3. Lead polls, sees `ASKING`, answers: `fleet_send{turnId: turnId1, content: "A1"}` →
`answer(turnId1, "A1", t)`. This opens a **new** waiter via `rendezvous.open(workerSession)`
(line 929) but — unlike `send()` — never puts it into `asyncTasksByWaiter`.
`answerAsk(turnId1, "A1")` unblocks the worker's `ask()` call.
`clearAsyncQuestion(turnId1, false)` clears `task1.question` but keeps
`asyncTasksByTurn[turnId1] = task1` (deliberate, per its own javadoc). `answer()` then blocks
on its own `reply.get()` (line 936).
4. Worker `W`, still in the same resumed turn, calls `fleet_ask{"Q2"}` again before replying →
a second `ask(W, "Q2", t)`. `openAsk` mints `turnId2`. `markAsyncQuestion` looks up
`asyncTasksByWaiter.get(waiter)` for the waiter `answer()` opened in step 3 — **not found**
(never registered), so `task == null`; `task1.turnId` stays `turnId1`, no
`asyncTasksByTurn[turnId2]` entry is ever created. `resolveQuestion` still succeeds (it only
needs a live waiter, not a `Task`) and wakes `answer(turnId1,...)`'s blocked `reply.get()`
with `Resolution(QUESTION, "Q2", turnId2)`.
5. `answer(turnId1,...)` (line 936-939): `result = Reply(Outcome.QUESTION, "Q2", turnId2)`;
`finishAsyncTask(turnId1, result)` looks up `task1` by the **original** `turnId1` (still
stamped from step 3) and unconditionally does `task1.future.complete(result)` — completing
`ticket1`'s future with a **QUESTION** outcome, then removes `asyncTasksByTurn[turnId1]`.
`answer()` returns `Outcome.QUESTION` to the lead's own `fleet_send{turnId1,...}` call
(correct, and separately answerable via `turnId2`), but `ticket1` is now terminally done.
### What goes wrong
- `fleet_poll{ticket1}` now hits the `f.isDone()` branch in `poll()` permanently. `Outcome.QUESTION`
is not `REPLIED`/`COMPLETED_UNREPLIED` (`r.completed()` is false) and carries no
`WORKER_FAILED`/`BACKEND_EXHAUSTED` reason, so it falls through to
`Phase.FAILED`, `detail = "no reply — question"` — even though the worker is alive and only
waiting on `turnId2`.
- `asyncTasksByTurn` no longer has any entry for `task1`/`target`, so
`hasAsyncQuestion(target)` goes back to `false` immediately, and `hasOrphanedDelegation`
no longer excludes this target's real state correctly either.
- If the worker's eventual real `fleet_reply` (after `turnId2` is answered, or times out and it
finishes on its own) is not captured by a chained direct `answer(turnId2,...)` call,
`reply()`'s fast path (`rendezvous.resolve`) finds no live waiter, `askAnsweredAsyncTasks`
finds no candidate (`task1.future.isDone()` is already true, so it is excluded), and the reply
is silently dropped into the inbox as a **stranded reply** — unreachable from `ticket1` and
from `abandon()`'s stranded-reply recovery (no open `matching` task exists any more).
### Confidence
High. I traced this with no races or interleavings assumed beyond the documented, deterministic
CB-205 chained-ask protocol that `outcomeOf`/`answer()` already generically support (mapping
`Kind.QUESTION` through `answer()`'s own return value is clearly intentional — see the
`Reply.turnId()` javadoc). I did not run the suite, but grepped
`src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java` for a test exercising a *second*
`fleet_ask` inside one resumed (answered) turn and found none — `asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter`
and `askSurfacesAsAQuestionAndTheAnswerResumesTheSameTurn` both cover only a single ask per
turn. A `git log -p` on this file also turned up the CB-588 comment (now at
`sendAsync`, describing the `whenComplete` hook) which explicitly frames
`answer()`'s `finishAsyncTask(turnId, result)` as firing "once a QUESTION is resolved" —
i.e. the author modeled that call as inherently terminal, which is exactly the assumption this
bug violates when the resumed turn asks again.
## Secondary (much shorter)
1. **Root-cause detail, same defect as above** — `answer()` (line 929) never puts its freshly
opened waiter into `asyncTasksByWaiter`, unlike `send()` (line 802). Even if the outcome
guard above is added, a chained second ask still can't be re-attached to `task1` via the
normal `markAsyncQuestion` path without also fixing this registration gap.
2. **Low / shape only** — `pruneTerminalTickets()` (line ~1092) is only invoked from inside
`sendAsync()`. A fleet whose sessions stop receiving new async sends (e.g. everything now
goes through blocking `send()`, or the target churns and workers are torn down) never prunes
its already-terminal `tasks` entries past `TICKET_TTL_NANOS`. Not reachable as a "stuck"
ticket (tickets still resolve correctly), only as unbounded `tasks`/`asyncTasksByWaiter`-adjacent
memory growth over a long-lived daemon with no further `sendAsync` traffic; did not verify
this is realistic in production traffic patterns, flagging as a shape only.
@@ -250,7 +250,9 @@ 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 -> liveSessionCount(sessions.roster(), profileName));
liveCountRef.set(profileName -> (int) sessions.roster().stream()
.filter(s -> profileName.equals(s.profile()))
.count());
// CB-303 part 1: idle-ttl reaper — only when configured, defaults to disabled.
final SessionReaper reaper;
@@ -857,19 +859,6 @@ 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,9 +322,8 @@ 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 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.
* instead, naming the profile and the key. Live means any session the
* registry still owns (acquired and not yet released), in any state.
* @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
@@ -28,14 +28,6 @@ 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;
@@ -179,7 +171,6 @@ public final class FleetHealthMonitor {
// member stayed terminal.
if (terminal(next)) {
failTerminalTarget(target, next);
scheduleTerminalRecheck(target, next);
}
}
@@ -200,48 +191,6 @@ 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, toolAction("fleet_send", req.arguments()),
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SEND,
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, toolAction("fleet_reply", req.arguments()), self);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.REPLY, 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, toolAction("fleet_ask", req.arguments()), self);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.ASK, 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, toolAction("fleet_status", req.arguments()), null);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, 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, toolAction("fleet_poll", a), target);
McpSchema.CallToolResult denied = deny(exchange, pollAction(target), 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, toolAction("fleet_ack", a), str(a, "target"));
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.DRAIN, 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, toolAction("fleet_spawn", req.arguments()), null);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SPAWN, 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, toolAction("fleet_list", Map.of()), null);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, 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, toolAction("fleet_stop", req.arguments()), paneId);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.STOP, paneId);
if (denied != null) return denied;
return stop(sessions, paneId);
};
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> profilesHandler =
(exchange, _) -> {
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_profiles", Map.of()), null);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
if (denied != null) return denied;
return profiles(workers, quarantine, outage);
};
BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> whoamiHandler =
(exchange, _) -> {
McpSchema.CallToolResult denied = deny(exchange, toolAction("fleet_whoami", Map.of()), null);
McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
if (denied != null) return denied;
return whoami(principal(exchange), sessions);
};
@@ -754,25 +754,6 @@ 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)) {
@@ -1167,35 +1148,15 @@ 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 reclaimable = reclaimable(session, messages);
boolean open = messages != null && messages.hasAcceptedDelivery(session.terminalId());
boolean inbox = messages != null && messages.hasInboxMessage(session.terminalId());
boolean reclaimable = (session.state() == MemberSession.State.READY || session.state() == MemberSession.State.DONE)
&& !open && !inbox;
row.put("reclaimable", reclaimable);
row.put("idleForSeconds", reclaimable ? Math.max(0, (nowNanos - session.lastActivityAtNanos()) / 1_000_000_000L) : null);
return row;
}
/**
* 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
@@ -1234,7 +1195,8 @@ 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 -> reclaimable(s, messages))
.filter(s -> (s.state() == MemberSession.State.READY || s.state() == MemberSession.State.DONE))
.filter(s -> messages == null || (!messages.hasAcceptedDelivery(s.terminalId()) && !messages.hasInboxMessage(s.terminalId())))
.count();
Map<String, Object> row = new LinkedHashMap<>();
row.put("profile", profile); row.put("maxLoad", cap); row.put("live", live);
@@ -540,66 +540,16 @@ 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} — refused
* outright when {@code memberHerdrSocket} is configured, see above
* {@code null}/blank to target the default {@code ~/.claude.json}
* @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 void seedTrustDialog(String configDir, String cwd) {
private static 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");
@@ -609,20 +559,18 @@ 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. (Reached only when
// memberHerdrSocket is absent — the block above already refused otherwise.)
// person's home config and deserves its own log line.
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 && !written; attempt++) {
for (int attempt = 1; attempt <= MAX_TRUST_JSON_CAS_ATTEMPTS; attempt++) {
byte[] before = Files.isRegularFile(target) ? Files.readAllBytes(target) : null;
ObjectNode root = parseTrustJsonOrEmpty(before);
JsonNode projectsNode = root.get("projects");
@@ -663,39 +611,26 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
continue;
}
writeAtomically(target, newContent);
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);
return;
}
// 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,33 +181,6 @@ 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) {
@@ -927,16 +927,7 @@ 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
}
@@ -944,21 +935,7 @@ 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());
// #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);
}
finishAsyncTask(turnId, result);
return result;
} catch (TimeoutException e) {
// The worker resumed but hasn't replied yet — no completion fallback arms an answered
@@ -971,7 +948,6 @@ 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 {
@@ -1130,16 +1106,9 @@ 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;
}
@@ -47,22 +47,6 @@ 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;
@@ -238,7 +222,7 @@ public final class FleetApp {
/** Prometheus scrape endpoint (CB-502). */
private void metrics(Context ctx) {
if (!allow(ctx, routeAction("GET /metrics"), null)) {
if (!allow(ctx, Authz.Action.METRICS, null)) {
return;
}
ctx.status(200).contentType("text/plain; version=0.0.4; charset=utf-8").result(metrics.render());
@@ -310,7 +294,7 @@ public final class FleetApp {
* member workspace (they live on the member daemon only).
*/
private void sessions(Context ctx) {
if (!allow(ctx, routeAction("GET /sessions"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
List<Map<String, Object>> out = new ArrayList<>();
@@ -335,7 +319,7 @@ public final class FleetApp {
/** Discovery: every agent herdr tracks, keyed by its Claude session UUID. */
private void agents(Context ctx) {
if (!allow(ctx, routeAction("GET /agents"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
ctx.status(200).json(Map.of("agents",
@@ -344,7 +328,7 @@ public final class FleetApp {
/** CB-304: bridge-owned roster merged with live herdr status by paneId. */
private void listMembers(Context ctx) {
if (!allow(ctx, routeAction("GET /members"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
// CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
@@ -375,7 +359,7 @@ public final class FleetApp {
/** The configured worker profiles and which one a no-argument spawn uses. */
private void profiles(Context ctx) {
if (!allow(ctx, routeAction("GET /profiles"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
ctx.status(200).json(Map.of(
@@ -392,7 +376,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, routeAction("GET /member-credentials"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
MemberCredentialPolicyView view = memberCredentials.get();
@@ -412,7 +396,7 @@ public final class FleetApp {
* the subscription boundary, 400 for an unknown profile.
*/
private void spawnMember(Context ctx) {
if (!allow(ctx, routeAction("POST /members"), null)) {
if (!allow(ctx, Authz.Action.SPAWN, null)) {
return;
}
String role = ctx.queryParam("role");
@@ -483,7 +467,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, routeAction("DELETE /members/{paneId}"), paneId)) {
if (!allow(ctx, Authz.Action.STOP, paneId)) {
return;
}
sessions.release(paneId);
@@ -498,7 +482,7 @@ public final class FleetApp {
*/
private void sendMessage(Context ctx) {
String id = ctx.pathParam("id");
if (!allow(ctx, routeAction("POST /sessions/{id}/message"), id)) {
if (!allow(ctx, Authz.Action.SEND, id)) {
return;
}
String content;
@@ -585,7 +569,7 @@ public final class FleetApp {
*/
private void askMessage(Context ctx) {
String id = ctx.pathParam("id");
if (!allow(ctx, routeAction("POST /sessions/{id}/ask"), id)) {
if (!allow(ctx, Authz.Action.ASK, id)) {
return;
}
String question;
@@ -624,7 +608,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, routeAction("POST /sessions/{id}/reply"), id)) {
if (!allow(ctx, Authz.Action.REPLY, id)) {
return;
}
String content;
@@ -645,7 +629,7 @@ public final class FleetApp {
*/
private void drainReplies(Context ctx) {
String id = ctx.pathParam("id");
if (!allow(ctx, routeAction("GET /sessions/{id}/replies"), id)) {
if (!allow(ctx, Authz.Action.DRAIN, id)) {
return;
}
var replies = messages.drainReplies(id);
@@ -663,7 +647,7 @@ public final class FleetApp {
*/
private void sessionStatus(Context ctx) {
String id = ctx.pathParam("id");
if (!allow(ctx, routeAction("GET /sessions/{id}/status"), id)) {
if (!allow(ctx, Authz.Action.READ, id)) {
return;
}
try {
@@ -688,7 +672,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, routeAction("GET /tasks/{ticket}"), null)) {
if (!allow(ctx, Authz.Action.READ, null)) {
return;
}
MessageService.TaskView v = messages.poll(ctx.pathParam("ticket"));
@@ -213,18 +213,13 @@ public final class GitWorktrees implements Worktrees {
worktreePath, cleanup.getMessage());
}
try {
deleteBranch(repoRoot, branch);
exec("git", "-C", repoRoot, "branch", "-D", 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,20 +334,7 @@ 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) {
// 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());
}
worktrees.remove(worktrees.repoRoot(removed.cwd()), removed.worktree());
}
}
@@ -517,26 +504,11 @@ 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;
}
@@ -11,15 +11,6 @@ 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,44 +226,4 @@ 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,9 +24,7 @@ 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;
@@ -436,120 +434,4 @@ 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;
}
}
@@ -24,13 +24,8 @@ 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.*;
@@ -49,9 +44,6 @@ 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;
@@ -209,42 +201,6 @@ 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,7 +31,6 @@ 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;
@@ -605,60 +604,6 @@ 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();
@@ -2742,126 +2742,4 @@ 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");
}
}
@@ -975,64 +975,6 @@ 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,7 +1,6 @@
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;
@@ -26,15 +25,8 @@ 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.*;
@@ -44,10 +36,6 @@ 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;
@@ -104,49 +92,6 @@ 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
@@ -17,9 +17,6 @@ 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) {
}
@@ -38,7 +35,6 @@ 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<>();
@@ -55,8 +51,6 @@ 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";
@@ -104,21 +98,6 @@ 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;
@@ -153,14 +132,6 @@ 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
@@ -175,9 +146,6 @@ 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) {
@@ -250,14 +218,6 @@ 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,7 +74,6 @@ 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;
@@ -115,11 +114,6 @@ 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) {
@@ -164,10 +158,6 @@ class SessionManagerTest {
return List.copyOf(removeCalls);
}
List<String> deleteBranchCalls() {
return List.copyOf(deleteBranchCalls);
}
List<String> snapshotCalls() {
return List.copyOf(snapshotCalls);
}
@@ -999,21 +989,8 @@ 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 reapIdleCountsAllThreeSessionsWhenOnlyItsWorktreeRemovalFails() {
void reapIdleSurvivesOneSessionThatFailsToRelease() {
long[] clock = {0};
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees();
@@ -1027,8 +1004,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 — fleetd #283 makes release() catch and log
// this itself, so it no longer propagates out of release() at all.
// 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.
worktrees.failRemoveFor(b.worktree());
LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
@@ -1049,16 +1026,14 @@ class SessionManagerTest {
.map(ILoggingEvent::getFormattedMessage)
.filter(m -> m.contains(b.paneId()))
.findFirst()
.orElse("no worktree-removal-failure WARN logged");
.orElse("no reap-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(3, reaped,
"fleetd #283: release() no longer throws for a worktree-removal failure, so reapIdle "
+ "counts all three idle sessions as reaped");
assertEquals(2, reaped, "the middle session's failure is logged, not counted 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(),
@@ -212,41 +212,9 @@ 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
@@ -384,39 +352,6 @@ 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