Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b8b25cf74c | |||
| 76672ff016 | |||
| 9379f92c23 | |||
| 21ff63b11d | |||
| 21844b54d7 | |||
| 4769481515 | |||
| bbbb4c1eb3 | |||
| 38c248e617 | |||
| 9debc0de27 | |||
| f34361b263 | |||
| 85c90d440a | |||
| e97502d550 | |||
| aef14ff46e | |||
| cba516bda4 |
@@ -625,6 +625,19 @@ public final class Fleetd {
|
||||
log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
|
||||
}
|
||||
|
||||
// fleetd #297: named once and reused verbatim below for FleetApp's GET /profiles, rather than
|
||||
// built a second time — two independently-constructed sources reading the SAME BackendQuarantine
|
||||
// / BackendOutagePolicy would still be able to drift (e.g. a future edit to the credentialIdFor
|
||||
// closure in only one of the two places), exactly the shape #284 was.
|
||||
FleetMcp.QuarantineSource quarantineSource = new FleetMcp.QuarantineSource(profile -> {
|
||||
var configured = config.get().profiles().get(profile);
|
||||
return configured == null ? null : configured.effectiveCredentialId();
|
||||
}, quarantine);
|
||||
FleetMcp.OutageSource outageSource = new FleetMcp.OutageSource(profile -> {
|
||||
var configured = config.get().profiles().get(profile);
|
||||
return configured == null ? null : configured.effectiveCredentialId();
|
||||
}, outagePolicy);
|
||||
|
||||
FleetMcp mcp = new FleetMcp(messages, workers, sessions, identity, presence,
|
||||
primaryRegistry, callers, metrics, new FleetMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
|
||||
profile -> {
|
||||
@@ -636,15 +649,9 @@ public final class Fleetd {
|
||||
return FleetHealthMonitor.coverage(health != null && health.isEnabled(),
|
||||
health != null && health.notifications() != null && health.notifications().configured());
|
||||
}),
|
||||
new FleetMcp.QuarantineSource(profile -> {
|
||||
var configured = config.get().profiles().get(profile);
|
||||
return configured == null ? null : configured.effectiveCredentialId();
|
||||
}, quarantine),
|
||||
quarantineSource,
|
||||
leadMailbox,
|
||||
new FleetMcp.OutageSource(profile -> {
|
||||
var configured = config.get().profiles().get(profile);
|
||||
return configured == null ? null : configured.effectiveCredentialId();
|
||||
}, outagePolicy),
|
||||
outageSource,
|
||||
new FleetMcp.LeadSeatSource(leadSeatLookup(() -> config.get().profiles(), leaders, leads)));
|
||||
|
||||
// CB-637: the receive half. Only constructed when a lead mailbox actually opened — with no
|
||||
@@ -715,9 +722,12 @@ public final class Fleetd {
|
||||
// GET /sessions must merge across both, or a down/unpolled member daemon is invisible.
|
||||
// fleetd #111: live (re-read-per-request) memberCredentials view for GET /member-credentials —
|
||||
// same hot-reload shape as the memberCredentials supplier passed to ClaudeCodeLauncher above.
|
||||
// fleetd #297: quarantineSource/outageSource are the SAME instances passed to FleetMcp above —
|
||||
// GET /profiles must report the identical quarantine/cool-off facts as fleet_profiles.
|
||||
Javalin app = new FleetApp(herdr, memberHerdr, workers, sessions, messages, presence, mcp.servlet(),
|
||||
callers, metrics, deliverable,
|
||||
() -> MemberCredentialPolicyView.of(config.get().memberCredentials())).build();
|
||||
() -> MemberCredentialPolicyView.of(config.get().memberCredentials()),
|
||||
quarantineSource, outageSource).build();
|
||||
app.start(cfg.bind().host(), cfg.bind().port());
|
||||
log.info("fleetd listening on {}:{}, herdr socket {}",
|
||||
cfg.bind().host(), cfg.bind().port(), socket);
|
||||
|
||||
@@ -262,11 +262,15 @@ public final class CallerResolver {
|
||||
return token.isEmpty() ? null : token;
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #305: delegates to {@link ConnectionIdentity#isLoopback}. This used to be a second,
|
||||
* independent copy of the same rule, and the two drifted: this one accepted all of
|
||||
* {@code 127.0.0.0/8}, {@code ConnectionIdentity}'s accepted only {@code 127.0.0.1}. A caller
|
||||
* from {@code 127.0.0.2} therefore had its identity skipped (so it had no terminal) and was
|
||||
* then read as loopback here — which under loopback-trust is the primary. Sharing the inputs
|
||||
* would not have prevented that; only sharing the computation does.
|
||||
*/
|
||||
private static boolean isLoopback(String remoteAddr) {
|
||||
if (remoteAddr == null) {
|
||||
return false;
|
||||
}
|
||||
return remoteAddr.equals("127.0.0.1") || remoteAddr.equals("::1")
|
||||
|| remoteAddr.equals("0:0:0:0:0:0:0:1") || remoteAddr.startsWith("127.");
|
||||
return ConnectionIdentity.isLoopback(remoteAddr);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -157,6 +157,7 @@ public final class Injector {
|
||||
boolean awaitingCompletion; // a delivered message's turn is not yet known-complete
|
||||
boolean turnObserved; // saw a real `working` sample since that delivery (turn ran)
|
||||
int unknownSinceTurn; // consecutive `unknown` samples while a delegation is outstanding (CB-109)
|
||||
int unknownSincePostTurn; // the same, for the post-turn housekeeping phase (fleetd #306)
|
||||
int notReadySincePoll; // consecutive injectable samples a queued message waited on the readiness gate (CB-114)
|
||||
boolean postTurnPending; // completion observed; adapter housekeeping has not started yet
|
||||
boolean awaitingPostTurnPickup;
|
||||
@@ -217,10 +218,12 @@ public final class Injector {
|
||||
t.awaitingPickup = false;
|
||||
t.injectableSincePickup = 0;
|
||||
t.unknownSinceTurn = 0;
|
||||
t.unknownSincePostTurn = 0;
|
||||
t.notReadySincePoll = 0;
|
||||
if (t.awaitingCompletion) t.turnObserved = true;
|
||||
} else if (status.injectable()) { // IDLE or BLOCKED
|
||||
t.unknownSinceTurn = 0;
|
||||
t.unknownSincePostTurn = 0;
|
||||
if (t.awaitingPostTurnPickup) {
|
||||
if (++t.injectableSincePostTurnPickup >= PICKUP_GRACE_POLLS) {
|
||||
t.awaitingPostTurnPickup = false;
|
||||
@@ -315,6 +318,24 @@ public final class Injector {
|
||||
t.unknownSinceTurn = 0;
|
||||
turnFailed = true;
|
||||
}
|
||||
// fleetd #306: the same escape for the post-turn housekeeping phase. Four latches
|
||||
// gate delivery (awaitingCompletion, postTurnPending, awaitingPostTurnPickup,
|
||||
// postTurnObserved) and only the first had a way out of a sustained unknown streak —
|
||||
// a gate that closed one direction only. The other two below are released here as
|
||||
// well; postTurnPending needs no escape because it is cleared unconditionally on the
|
||||
// line after the listener call that sets it.
|
||||
//
|
||||
// This does NOT set turnFailed. The delegated turn already completed and its waiter
|
||||
// already resolved — what is outstanding is adapter housekeeping (the `/clear`).
|
||||
// Reporting a turn failure here would drive SessionManager.onFailed on a session
|
||||
// that genuinely finished its work, which is a worse lie than the wedge.
|
||||
if ((t.awaitingPostTurnPickup || t.postTurnObserved)
|
||||
&& ++t.unknownSincePostTurn >= TURN_STALL_GRACE_POLLS) {
|
||||
t.awaitingPostTurnPickup = false;
|
||||
t.postTurnObserved = false;
|
||||
t.injectableSincePostTurnPickup = 0;
|
||||
t.unknownSincePostTurn = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Reclaim the entry once the worker is fully quiescent (nothing queued, no pickup or
|
||||
|
||||
@@ -60,7 +60,30 @@ public final class ConnectionIdentity {
|
||||
return pid > 0 ? cwds.cwdForPid(pid) : null;
|
||||
}
|
||||
|
||||
private static boolean isLoopback(String addr) {
|
||||
return "127.0.0.1".equals(addr) || "::1".equals(addr) || "0:0:0:0:0:0:0:1".equals(addr);
|
||||
/**
|
||||
* Whether {@code addr} is a same-host address, and therefore one whose peer PID is worth
|
||||
* looking up. <strong>This is the one definition of loopback in the daemon</strong> —
|
||||
* {@code CallerResolver} calls it rather than keeping its own, because the two used to differ
|
||||
* and that difference was a privilege escalation (fleetd #305).
|
||||
*
|
||||
* <p>The whole of {@code 127.0.0.0/8} counts, not just {@code 127.0.0.1}. On Linux every
|
||||
* address in that range is bound to {@code lo} by default, so a process can connect to
|
||||
* {@code 127.0.0.1:8765} with a source address of {@code 127.0.0.2} — measured on the Linux
|
||||
* fleet host, where binding that source succeeds.
|
||||
*
|
||||
* <p><strong>Being strict here does not make the daemon safer; it makes it unsafe.</strong>
|
||||
* That reads backwards, so it is worth stating plainly. This predicate does not decide whether
|
||||
* a caller is trusted — it decides whether the caller's identity is <em>resolved at all</em>.
|
||||
* Returning false means {@link #resolve} answers "no terminal", and downstream a caller with no
|
||||
* terminal is treated as the primary under loopback-trust. So every address excluded here is an
|
||||
* address on which a worker silently becomes the lead. Widening a check normally weakens it;
|
||||
* widening this one is what closes the hole.
|
||||
*/
|
||||
public static boolean isLoopback(String addr) {
|
||||
if (addr == null) {
|
||||
return false;
|
||||
}
|
||||
String a = addr.startsWith("::ffff:") ? addr.substring(7) : addr; // IPv4-mapped IPv6
|
||||
return a.startsWith("127.") || "::1".equals(a) || "0:0:0:0:0:0:0:1".equals(a);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -813,7 +813,12 @@ public final class FleetMcp {
|
||||
return error("fleet_reply is for workers only — could not identify the calling worker "
|
||||
+ "from the connection");
|
||||
}
|
||||
if (content == null) {
|
||||
// fleetd #302: isBlank, not == null, to match fleet_send's own guard above. MessageService
|
||||
// .reply now REJECTS blank content, and this handler is a bare BiFunction with no try/catch
|
||||
// around it — so a whitespace-only fleet_reply would leave here as an uncaught
|
||||
// IllegalArgumentException instead of this clean tool error. Null and whitespace are the
|
||||
// same mistake by the caller and must get the same answer.
|
||||
if (isBlank(content)) {
|
||||
return error("content is required");
|
||||
}
|
||||
messages.reply(callerTerminal, content);
|
||||
@@ -1015,6 +1020,25 @@ public final class FleetMcp {
|
||||
* both maps at once when it is both exhaustion-quarantined AND cooling off.
|
||||
*/
|
||||
static McpSchema.CallToolResult profiles(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
|
||||
return text(json(profilesView(workers, quarantine, outage)));
|
||||
}
|
||||
|
||||
/**
|
||||
* The body both front doors answer {@code profiles} with: the configured profile names, the
|
||||
* default, and the two independent outage states — {@code quarantined} (the backend reported it
|
||||
* out of capacity) and {@code coolingOff} (the credential threw repeated non-exhaustion backend
|
||||
* errors). Each map is present only when at least one profile is in that state, and a profile
|
||||
* can appear in both at once, because the two checks are separate.
|
||||
*
|
||||
* <p>fleetd #297: extracted so {@code fleet_profiles} and {@code GET /profiles} render from ONE
|
||||
* body builder rather than two copies. Passing both doors the same {@link QuarantineSource} and
|
||||
* {@link OutageSource} instances is necessary but not sufficient: with the loop written out
|
||||
* twice, a later edit to the row shape — a renamed key, an added field — lands on one door and
|
||||
* not the other, and the two then disagree about a live outage. That is exactly what fleetd
|
||||
* #284 was, where one rule computed in two places was widened in only one and a single response
|
||||
* contradicted itself. Shared inputs do not make duplicated computation safe.
|
||||
*/
|
||||
public static Map<String, Object> profilesView(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
|
||||
Map<String, Object> result = new LinkedHashMap<>();
|
||||
result.put("profiles", workers.profiles());
|
||||
result.put("default", workers.defaultProfile() == null ? "" : workers.defaultProfile());
|
||||
@@ -1046,7 +1070,7 @@ public final class FleetMcp {
|
||||
if (!coolingOff.isEmpty()) {
|
||||
result.put("coolingOff", coolingOff);
|
||||
}
|
||||
return text(json(result));
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -697,7 +697,20 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
if (paneId == null) {
|
||||
throw new IllegalStateException("pane.split returned no pane — cannot start a peer");
|
||||
}
|
||||
Agent peer = startUniquelyNamed(cfg, argv, paneId).agent();
|
||||
Agent peer;
|
||||
try {
|
||||
peer = startUniquelyNamed(cfg, argv, paneId).agent();
|
||||
} catch (RuntimeException e) {
|
||||
// The peer never started — don't leave the pane we just created orphaned.
|
||||
// Best-effort cleanup; never let it mask the real spawn failure.
|
||||
try {
|
||||
stop(paneId);
|
||||
} catch (RuntimeException cleanup) {
|
||||
log.warn("failed to close orphaned pane {} after spawn error: {}",
|
||||
paneId, cleanup.getMessage());
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
log.info("{} started pane={} terminal={}", namePrefix, peer.paneId(), peer.terminalId());
|
||||
return peer;
|
||||
}
|
||||
@@ -996,7 +1009,8 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
* that gap: it stops waiting immediately (never burns the rest of the timeout), runs the same
|
||||
* teardown the timeout path below runs, and throws with a message that says the backend exited
|
||||
* rather than that the pane was slow. Any other {@link HerdrException} still propagates
|
||||
* unchanged — this gate does not know how to recover from it.
|
||||
* unchanged — this gate does not interpret or recover from it, but it still closes the pane
|
||||
* it opened before handing the exception to its caller.
|
||||
*/
|
||||
private void waitUntilInjectableOrThrow(String paneId) {
|
||||
long start = nowMillis.getAsLong();
|
||||
@@ -1010,7 +1024,15 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
if (isAlreadyGone(e)) {
|
||||
failFastOnGoneBackend(paneId, e, nowMillis.getAsLong() - start);
|
||||
}
|
||||
throw e; // any other herdr failure is not ours to interpret — let it propagate
|
||||
// This gate must not interpret an unrelated herdr error, but the caller does not
|
||||
// receive paneId when spawn throws. Close the pane here before propagating e unchanged.
|
||||
try {
|
||||
stop(paneId);
|
||||
} catch (RuntimeException cleanup) {
|
||||
log.warn("failed to close orphaned pane {} after readiness-gate error: {}",
|
||||
paneId, cleanup.getMessage());
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
lastStatus = sample.status();
|
||||
if (lastStatus.injectable() || refinedInjectable(paneId, sample)) {
|
||||
|
||||
@@ -187,6 +187,24 @@ public final class MessageService {
|
||||
private volatile Long completedNanos;
|
||||
private volatile Reply question;
|
||||
private volatile String turnId;
|
||||
/**
|
||||
* Set when this task's {@code fleet_ask} lapsed with no answer (fleetd #307):
|
||||
* {@link #clearAsyncQuestion} then forgets {@link #turnId} (nulls it and drops the task from
|
||||
* {@code asyncTasksByTurn}) so {@link #hasAsyncQuestion} stops reporting the target BUSY — a
|
||||
* later {@code fleet_send} to it must be accepted, not refused. But the worker's turn is
|
||||
* still genuinely live: it resumed on its own and will eventually call its real
|
||||
* {@code fleet_reply}. Losing {@link #turnId} loses {@link #askAnsweredAsyncTasks}' only
|
||||
* signal that such a reply belongs to this task, so that reply used to fall straight to the
|
||||
* inbox and strand — {@code fleet_poll} stayed {@code PENDING} forever, later force-failed by
|
||||
* {@link #abandon} with the misleading "session released before it replied". This flag is a
|
||||
* second, independent signal that survives the forgetting: {@link #askAnsweredAsyncTasks}
|
||||
* accepts it in place of a live {@link #turnId}, without ever re-adding the task to
|
||||
* {@code asyncTasksByTurn} (so the BUSY release is untouched). Cleared implicitly once
|
||||
* {@link #future} resolves — every match in {@link #askAnsweredAsyncTasks} already requires
|
||||
* {@code !future.isDone()}, so a task that recovered (or was later failed by
|
||||
* {@link #abandon}) can never match again regardless of this flag's value.
|
||||
*/
|
||||
private volatile boolean askTimedOut;
|
||||
|
||||
private Task(String ticket, String target, LongSupplier nowNanos) {
|
||||
this.ticket = ticket;
|
||||
@@ -395,36 +413,55 @@ public final class MessageService {
|
||||
* {@link Rendezvous#resolveQuestion} must keep today's {@code NO_WAITER} behaviour — questions
|
||||
* are interactive and must never be queued.
|
||||
*
|
||||
* <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks}
|
||||
* cannot actually return more than one entry today (see its own javadoc for why — in short,
|
||||
* {@link #hasAsyncQuestion} keeps a target BUSY, so no second task can reach this state, for as
|
||||
* long as an earlier one's {@code turnId} is still stamped). That is an emergent guarantee from
|
||||
* two other facts, not one this method enforces, so this branch stays in as defence in depth
|
||||
* rather than being removed as dead code: if it ever weakens, returning whichever candidate a
|
||||
* {@code ConcurrentHashMap} iteration reaches first would let a genuine reply complete the
|
||||
* <em>wrong</em> ticket — silently handing the lead something that reads like a correct answer to
|
||||
* a delegation the worker never touched, which is worse than a failure because the lead acts on
|
||||
* it. When more than one candidate exists, guessing is not safe: fall back to the inbox exactly
|
||||
* as the zero-candidate case does, and let {@link #abandon} apply the eventual recovery
|
||||
* deterministically instead.
|
||||
* <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks} can
|
||||
* return more than one entry — a reachable state, not a hypothetical one (see its own javadoc:
|
||||
* an {@code fleet_ask} that lapsed with no answer, fleetd #307, frees the target for a completely fresh
|
||||
* delegation, which can itself go on to ask-and-lapse before the first worker's real reply
|
||||
* arrives). Returning whichever candidate a {@code ConcurrentHashMap} iteration reaches first
|
||||
* would let a genuine reply complete the <em>wrong</em> ticket — silently handing the lead
|
||||
* something that reads like a correct answer to a delegation the worker never touched, which is
|
||||
* worse than a failure because the lead acts on it. When more than one candidate exists, guessing
|
||||
* is not safe: fall back to the inbox exactly 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
|
||||
}
|
||||
// #137: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming a
|
||||
// turn that {@link #answer} already gave up waiting on. answer()'s own bounded wait (the
|
||||
// primary's fleet_send{turnId} call, capped well under a minute) can time out and close its
|
||||
// waiter long before the worker — now actually resuming real work — finishes and replies. That
|
||||
// reply used to have nowhere to land but the session inbox, leaving the async ticket's future
|
||||
// unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's abandon() forced it
|
||||
// FAILED with a misleading "session released before it replied" reason, even though the reply
|
||||
// had, in fact, arrived. Completing the matching ticket directly here means fleet_poll{ticket}
|
||||
// sees the real reply instead.
|
||||
// #137/fleetd #307: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming
|
||||
// a turn that either answer() (#137) or ask() (fleetd #307) already gave up waiting on:
|
||||
// - answer()'s own bounded wait (the primary's fleet_send{turnId} call, capped well under a
|
||||
// minute) can time out and close its waiter long before the worker — now actually resuming
|
||||
// real work — finishes and replies.
|
||||
// - ask()'s own wait for the primary can time out first, with the worker resuming on its own
|
||||
// and finishing unanswered.
|
||||
// Either way that reply used to have nowhere to land but the session inbox, leaving the async
|
||||
// ticket's future unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's
|
||||
// abandon() forced it FAILED with a misleading "session released before it replied" reason,
|
||||
// even though the reply had, in fact, arrived. Completing the matching ticket directly here
|
||||
// means fleet_poll{ticket} sees the real reply instead.
|
||||
List<Task> candidates = askAnsweredAsyncTasks(session);
|
||||
if (candidates.size() == 1) {
|
||||
Task orphan = candidates.get(0);
|
||||
@@ -455,34 +492,42 @@ public final class MessageService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Every still-open async task on {@code target} whose {@code fleet_ask} was already answered —
|
||||
* its {@link Task#turnId} is stamped but its {@link Task#question} was cleared by {@link #answer}
|
||||
* — yet whose future is not resolved yet (#137). Empty if no such task exists, including the
|
||||
* common case where {@code target}'s worker never used {@code fleet_ask} at all (a task that was
|
||||
* never asked has {@code turnId == null}, so it can never match here and only ever completes
|
||||
* through the ordinary rendezvous fast path in {@link #reply}).
|
||||
* Every still-open async task on {@code target} whose worker is genuinely expected to send a
|
||||
* real {@code fleet_reply} next with nothing left registered to catch it: either its
|
||||
* {@code fleet_ask} was already answered — {@link Task#turnId} is stamped but {@link
|
||||
* Task#question} was cleared by {@link #answer} — or its {@code fleet_ask} lapsed unanswered and
|
||||
* {@link Task#askTimedOut} marks that (fleetd #307; {@link Task#turnId} is {@code null} by then, forgotten
|
||||
* so the target is not left BUSY — see {@link Task#askTimedOut}'s own javadoc). Either way the
|
||||
* task's future is not resolved yet. Empty if no such task exists, including the common case
|
||||
* where {@code target}'s worker never used {@code fleet_ask} at all (a task that was never asked
|
||||
* has both {@code turnId == null} and {@code askTimedOut == false}, so it can never match here and
|
||||
* only ever completes through the ordinary rendezvous fast path in {@link #reply}).
|
||||
*
|
||||
* <p><strong>Returns at most one entry today — verified, not assumed.</strong> {@link #send}
|
||||
* refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is true, and that
|
||||
* check matches ANY task whose {@code turnId} is still stamped in {@code asyncTasksByTurn} —
|
||||
* not only while its question is still open. {@link #answer} deliberately leaves that stamp in
|
||||
* place ({@code clearAsyncQuestion(turnId, false)}) until the resumed turn's own future actually
|
||||
* resolves, at which point {@link #finishAsyncTask} both removes the stamp AND completes that
|
||||
* task's future in the same call. So a second task can never reach "{@code turnId} stamped, future
|
||||
* still open" — the exact pair this method matches on — while a first one already holds it: by
|
||||
* the time the stamp is gone, so is the eligibility. This is an emergent property of those two
|
||||
* facts holding together, not something this method (or its callers) enforces on its own — flip
|
||||
* {@code forgetTurn} to {@code true} in that one {@link #answer} call and it silently stops being
|
||||
* true, with nothing left to fail loudly. The callers below still handle "more than one" as
|
||||
* defence in depth against exactly that, not because they exercise it today: {@link #reply}
|
||||
* treats it as unresolvable and falls back to the inbox; {@link #abandon} would pick the oldest
|
||||
* deterministically (its own {@code matching} list has no such guarantee — see its javadoc).
|
||||
* <p><strong>Can return more than one entry — reachable, not just defence in depth.</strong>
|
||||
* {@link #send} refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is
|
||||
* true, and that check matches ANY task whose {@code turnId} is still stamped in
|
||||
* {@code asyncTasksByTurn}. While a task's {@code turnId} stays stamped — {@link #answer} leaves
|
||||
* it in place ({@code clearAsyncQuestion(turnId, false)}) until {@link #finishAsyncTask} removes
|
||||
* the stamp and completes the future in the same call — no second task on the same target can
|
||||
* reach an eligible state, because {@link #send} would refuse it as BUSY first. That single-task
|
||||
* guarantee holds only for the {@code turnId}-stamped half of this method's match: an
|
||||
* {@link Task#askTimedOut} task is, by construction, no longer stamped in {@code asyncTasksByTurn}
|
||||
* (that is the whole point of forgetting {@code turnId} in {@link #clearAsyncQuestion}), so the
|
||||
* target is free the moment one ask lapses. A fresh, independent {@code sendAsync} to the same
|
||||
* target can then be dispatched, itself pause on {@code fleet_ask}, and itself time out — landing
|
||||
* a second {@code askTimedOut} task on the very target the first one is still waiting to answer
|
||||
* for. Two (or more) genuinely open tasks on one target is therefore a real, reachable state
|
||||
* today, not a hypothetical: {@link #reply} treats it as unresolvable and falls back to the
|
||||
* inbox rather than guess which task a reply belongs to (guessing wrong would hand the lead a
|
||||
* plausible-looking answer to a delegation the worker never touched — worse than a failure,
|
||||
* because the lead acts on it); {@link #abandon} instead picks the oldest deterministically (its
|
||||
* own {@code matching} list has a different, wider match — see its javadoc).
|
||||
*/
|
||||
private List<Task> askAnsweredAsyncTasks(String target) {
|
||||
List<Task> candidates = new ArrayList<>();
|
||||
for (Task task : tasks.values()) {
|
||||
if (target.equals(task.target) && task.question == null && task.turnId != null
|
||||
&& !task.future.isDone()) {
|
||||
if (target.equals(task.target) && task.question == null && !task.future.isDone()
|
||||
&& (task.turnId != null || task.askTimedOut)) {
|
||||
candidates.add(task);
|
||||
}
|
||||
}
|
||||
@@ -880,6 +925,12 @@ public final class MessageService {
|
||||
return new AskResult(AskOutcome.ANSWERED, answer);
|
||||
} catch (TimeoutException e) {
|
||||
log.debug("fleet_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
|
||||
// fleetd #307: mark the task BEFORE clearAsyncQuestion(forgetTurn=true) below drops it out of
|
||||
// asyncTasksByTurn and nulls its turnId — that forgetting is deliberate and stays (it is
|
||||
// what keeps the target from staying BUSY forever), but it would otherwise also erase
|
||||
// askAnsweredAsyncTasks' only signal that the worker's eventual real fleet_reply still
|
||||
// belongs to this task, stranding it in the inbox with a false "never replied" verdict.
|
||||
markAskTimedOut(ticket.turnId());
|
||||
clearAsyncQuestion(ticket.turnId(), true);
|
||||
return new AskResult(AskOutcome.TIMED_OUT, null);
|
||||
} catch (ExecutionException e) {
|
||||
@@ -1144,6 +1195,23 @@ public final class MessageService {
|
||||
return task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark {@code turnId}'s task as having a {@code fleet_ask} that lapsed with no answer (fleetd #307), so
|
||||
* {@link #askAnsweredAsyncTasks} still recognizes the worker's eventual real {@code fleet_reply}
|
||||
* as belonging to it after {@link #clearAsyncQuestion}'s {@code forgetTurn=true} erases
|
||||
* {@link Task#turnId} — see {@link Task#askTimedOut}. Must be called before that forgetting, while
|
||||
* {@code turnId} can still resolve the task in {@code asyncTasksByTurn}; a lookup afterward would
|
||||
* find nothing. Only when it matches the task's current turn — same guard as
|
||||
* {@link #clearAsyncQuestion} — so a chained second {@code fleet_ask} (#282) that already moved
|
||||
* the task to a fresh {@code turnId} cannot mark it for a turn that is no longer its own.
|
||||
*/
|
||||
private void markAskTimedOut(String turnId) {
|
||||
Task task = asyncTasksByTurn.get(turnId);
|
||||
if (task != null && turnId.equals(task.turnId)) {
|
||||
task.askTimedOut = true;
|
||||
}
|
||||
}
|
||||
|
||||
/** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
|
||||
private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
|
||||
// CB-582: tell the push loop first — like ticketCollected, a removal for a turnId it never
|
||||
|
||||
@@ -9,6 +9,7 @@ import dev.ltms.fleet.auth.Principal;
|
||||
import dev.ltms.fleet.guard.GuardException;
|
||||
import dev.ltms.fleet.metrics.FleetMetrics;
|
||||
import dev.ltms.fleet.metrics.Metrics;
|
||||
import dev.ltms.fleet.mcp.FleetMcp;
|
||||
import dev.ltms.fleet.herdr.Agent;
|
||||
import dev.ltms.fleet.herdr.HerdrClient;
|
||||
import dev.ltms.fleet.herdr.HerdrException;
|
||||
@@ -86,6 +87,12 @@ public final class FleetApp {
|
||||
// absent() (the honest "no policy configured" view) for every constructor that does not wire
|
||||
// a real one, so existing legacy call sites keep building without knowing this field exists.
|
||||
private final Supplier<MemberCredentialPolicyView> memberCredentials;
|
||||
// fleetd #297: the SAME shared sources FleetMcp.profiles/fleet_profiles reads (BackendQuarantine
|
||||
// and BackendOutagePolicy are each one instance for the whole daemon — see Fleetd wiring) so
|
||||
// GET /profiles cannot drift from fleet_profiles about which profile is quarantined/cooling off.
|
||||
// .none() (the honest "feature not wired" view) for every constructor that does not pass one.
|
||||
private final FleetMcp.QuarantineSource quarantine;
|
||||
private final FleetMcp.OutageSource outage;
|
||||
private final ObjectMapper mapper = new ObjectMapper();
|
||||
|
||||
/**
|
||||
@@ -146,6 +153,23 @@ public final class FleetApp {
|
||||
MessageService messages, MemberPresence presence,
|
||||
HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
|
||||
Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials) {
|
||||
this(herdr, memberHerdr, workers, sessions, messages, presence, mcpServlet, auth, metrics,
|
||||
deliverable, memberCredentials, FleetMcp.QuarantineSource.none(), FleetMcp.OutageSource.none());
|
||||
}
|
||||
|
||||
/**
|
||||
* @param quarantine the SAME {@link FleetMcp.QuarantineSource} instance passed to {@code
|
||||
* FleetMcp} (fleetd #297), so {@code GET /profiles} reports the identical
|
||||
* exhaustion-quarantine facts as {@code fleet_profiles} rather than a second,
|
||||
* independently-computed copy
|
||||
* @param outage the SAME {@link FleetMcp.OutageSource} instance passed to {@code FleetMcp} —
|
||||
* see {@code quarantine}; a SEPARATE check from it, never merged in
|
||||
*/
|
||||
public FleetApp(HerdrClient herdr, HerdrClient memberHerdr, PeerLauncher workers, SessionManager sessions,
|
||||
MessageService messages, MemberPresence presence,
|
||||
HttpServlet mcpServlet, CallerResolver auth, Metrics metrics,
|
||||
Predicate<String> deliverable, Supplier<MemberCredentialPolicyView> memberCredentials,
|
||||
FleetMcp.QuarantineSource quarantine, FleetMcp.OutageSource outage) {
|
||||
this.herdr = herdr;
|
||||
this.memberHerdr = memberHerdr != null ? memberHerdr : herdr;
|
||||
this.workers = workers;
|
||||
@@ -156,6 +180,8 @@ public final class FleetApp {
|
||||
this.auth = auth;
|
||||
this.metrics = metrics;
|
||||
this.memberCredentials = memberCredentials != null ? memberCredentials : MemberCredentialPolicyView::absent;
|
||||
this.quarantine = quarantine != null ? quarantine : FleetMcp.QuarantineSource.none();
|
||||
this.outage = outage != null ? outage : FleetMcp.OutageSource.none();
|
||||
}
|
||||
|
||||
/** Wire routes onto a fresh, unstarted Javalin instance. Caller starts it. */
|
||||
@@ -338,8 +364,15 @@ public final class FleetApp {
|
||||
if (!allow(ctx, routeAction("GET /agents"), null)) {
|
||||
return;
|
||||
}
|
||||
ctx.status(200).json(Map.of("agents",
|
||||
workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
|
||||
try {
|
||||
ctx.status(200).json(Map.of("agents",
|
||||
workers.list().stream().map(Agent.class::cast).map(FleetApp::view).toList()));
|
||||
} catch (HerdrException e) {
|
||||
// fleetd #297: workers.list() reaches herdr — a transport failure must land in the same
|
||||
// {error, detail} envelope every other failure path here uses, not escape as a bare
|
||||
// exception and leave Javalin's default handling to respond outside the JSON contract.
|
||||
herdrError(ctx, e);
|
||||
}
|
||||
}
|
||||
|
||||
/** CB-304: bridge-owned roster merged with live herdr status by paneId. */
|
||||
@@ -347,40 +380,57 @@ public final class FleetApp {
|
||||
if (!allow(ctx, routeAction("GET /members"), null)) {
|
||||
return;
|
||||
}
|
||||
// CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
|
||||
Map<String, Agent> live = workers.list().stream()
|
||||
.map(Agent.class::cast)
|
||||
.filter(a -> a.terminalId() != null)
|
||||
.collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
|
||||
// fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
|
||||
// uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
|
||||
List<Map<String, Object>> out = sessions.rosterResolved().stream()
|
||||
.map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
|
||||
.toList();
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
// fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
|
||||
// "workers", so a caller that read "members" saw an empty fleet and reported no members at
|
||||
// all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
|
||||
// REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
|
||||
// drops reads this endpoint. Drop the alias once nothing reads it.
|
||||
body.put("members", out);
|
||||
body.put("workers", out);
|
||||
// CB-586: operator visibility for the refs/wip snapshot store without shelling into the
|
||||
// repo — how many snapshot refs exist and roughly what they cost. Present only once a
|
||||
// worktree session has established the repo, so a never-snapshotted fleet reports nothing.
|
||||
sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
|
||||
Map.of("count", st.count(), "costBytes", st.costBytes())));
|
||||
ctx.status(200).json(body);
|
||||
try {
|
||||
// CB-519: the registry key is a host-unique id, not the pane coordinate — join on terminal.
|
||||
Map<String, Agent> live = workers.list().stream()
|
||||
.map(Agent.class::cast)
|
||||
.filter(a -> a.terminalId() != null)
|
||||
.collect(Collectors.toMap(Agent::terminalId, Function.identity(), (_, b) -> b));
|
||||
// fleetd #209: this REST roster reports agentSessionId via SessionManager.rosterView, so it
|
||||
// uses the resolving roster read (caller-driven, not a timer) rather than the plain one.
|
||||
List<Map<String, Object>> out = sessions.rosterResolved().stream()
|
||||
.map(s -> SessionManager.rosterView(s, live.get(s.terminalId())))
|
||||
.toList();
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
// fleetd #199: the endpoint became /members in the CB-634 rename but the body key stayed
|
||||
// "workers", so a caller that read "members" saw an empty fleet and reported no members at
|
||||
// all. "members" is the canonical key; "workers" stays as a deprecated alias so an existing
|
||||
// REST consumer keeps working — the out-of-band path a lead falls back to when its MCP mount
|
||||
// drops reads this endpoint. Drop the alias once nothing reads it.
|
||||
body.put("members", out);
|
||||
body.put("workers", out);
|
||||
// CB-586: operator visibility for the refs/wip snapshot store without shelling into the
|
||||
// repo — how many snapshot refs exist and roughly what they cost. Present only once a
|
||||
// worktree session has established the repo, so a never-snapshotted fleet reports nothing.
|
||||
sessions.wipRefs().ifPresent(st -> body.put("wipRefs",
|
||||
Map.of("count", st.count(), "costBytes", st.costBytes())));
|
||||
ctx.status(200).json(body);
|
||||
} catch (HerdrException e) {
|
||||
// fleetd #297: same reasoning as agents() above — this is the out-of-band roster a lead
|
||||
// falls back to when its MCP mount drops, so it must stay inside the JSON error contract
|
||||
// exactly when herdr is briefly unreachable, not escape as a bare exception.
|
||||
herdrError(ctx, e);
|
||||
}
|
||||
}
|
||||
|
||||
/** The configured worker profiles and which one a no-argument spawn uses. */
|
||||
/**
|
||||
* The configured worker profiles, which one a no-argument spawn uses, and (fleetd #297) the two
|
||||
* outage states {@code fleet_profiles} already reports: {@code quarantined} (CB-578 stage B —
|
||||
* the backend reported it out of capacity) and {@code coolingOff} (fleetd #201 Unit 5 — the
|
||||
* credential threw repeated non-exhaustion backend errors). Both are read from the SAME shared
|
||||
* {@link FleetMcp.QuarantineSource}/{@link FleetMcp.OutageSource} instances {@code FleetMcp}
|
||||
* reads, never recomputed, so the two doors cannot disagree about which profile is down and why.
|
||||
* Independent checks, so a profile can appear in both maps at once; each map is present only
|
||||
* when at least one profile is in that state.
|
||||
*/
|
||||
private void profiles(Context ctx) {
|
||||
if (!allow(ctx, routeAction("GET /profiles"), null)) {
|
||||
return;
|
||||
}
|
||||
ctx.status(200).json(Map.of(
|
||||
"profiles", workers.profiles(),
|
||||
"default", workers.defaultProfile() == null ? "" : workers.defaultProfile()));
|
||||
// fleetd #297: ONE body builder, shared with fleet_profiles. Handing both doors the same
|
||||
// QuarantineSource/OutageSource instances stops them reading different facts; rendering
|
||||
// through the same method stops them reporting those facts differently. Both are needed.
|
||||
ctx.status(200).json(FleetMcp.profilesView(workers, quarantine, outage));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -460,6 +510,12 @@ public final class FleetApp {
|
||||
ctx.status(400).json(Map.of("error", "unknown_profile", "detail", e.getMessage()));
|
||||
} catch (PeerUnreachableException e) {
|
||||
ctx.status(502).json(Map.of("error", "spawn_timeout", "detail", e.getMessage()));
|
||||
} catch (HerdrException e) {
|
||||
// fleetd #304: not every herdr failure on the spawn path is a readiness timeout, so
|
||||
// PeerUnreachableException above does not cover this. Without this catch the exception
|
||||
// escapes to Javalin's default 500, while fleet_spawn reports the same failure as a
|
||||
// clean named error (FleetMcp.spawn) — the #297 one-door-guarded shape.
|
||||
herdrError(ctx, e);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -480,13 +536,29 @@ public final class FleetApp {
|
||||
return (s == null || s.isBlank()) ? null : s;
|
||||
}
|
||||
|
||||
/** Tear a worker down by pane id. */
|
||||
/**
|
||||
* Tear a worker down by pane id.
|
||||
*
|
||||
* <p>fleetd #304: the {@code HerdrException} catch is not cosmetic. {@code release} deregisters
|
||||
* the session, notifies the release listener and preserves a dirty worktree <em>before</em> it
|
||||
* calls {@code launcher.stop}, so a throw from that stop arrives after the teardown the caller
|
||||
* asked for has already happened. Letting it escape gave Javalin's default 500, which tells the
|
||||
* caller to retry — and the retry finds nothing in the registry, reaches the same stop, and
|
||||
* throws again, so it can never succeed. {@code herdrError} instead answers 404 ("the pane is
|
||||
* gone, stop retrying") or 502 ("herdr is upstream and broken, a retry may help"), matching what
|
||||
* {@code fleet_stop} reports for the same failure.
|
||||
*/
|
||||
private void stopMember(Context ctx) {
|
||||
String paneId = ctx.pathParam("paneId");
|
||||
if (!allow(ctx, routeAction("DELETE /members/{paneId}"), paneId)) {
|
||||
return;
|
||||
}
|
||||
sessions.release(paneId);
|
||||
try {
|
||||
sessions.release(paneId);
|
||||
} catch (HerdrException e) {
|
||||
herdrError(ctx, e);
|
||||
return;
|
||||
}
|
||||
ctx.status(204);
|
||||
}
|
||||
|
||||
@@ -634,7 +706,19 @@ public final class FleetApp {
|
||||
ctx.status(400).json(Map.of("error", "bad_request", "detail", "body must be JSON"));
|
||||
return;
|
||||
}
|
||||
messages.reply(id, content);
|
||||
// fleetd #302: content is required. `.path("content").asText("")` above turns a missing key
|
||||
// into "" rather than throwing, so without this check an empty/blank reply used to reach
|
||||
// messages.reply(...) and silently resolve the lead's waiter — the same class of bug as the
|
||||
// sibling "content is required" guards on sendMessage/askMessage below, except this one wrote
|
||||
// a WRONG value instead of failing loudly. The check lives in MessageService.reply so both
|
||||
// this door and FleetMcp.reply inherit the same rule; this catch only translates it into the
|
||||
// {error, detail} envelope this file uses everywhere else.
|
||||
try {
|
||||
messages.reply(id, content);
|
||||
} catch (IllegalArgumentException e) {
|
||||
ctx.status(400).json(Map.of("error", "bad_request", "detail", e.getMessage()));
|
||||
return;
|
||||
}
|
||||
ctx.status(200).json(Map.of("sessionId", id, "delivered", true));
|
||||
}
|
||||
|
||||
|
||||
@@ -864,10 +864,20 @@ public final class SessionManager implements TurnListener {
|
||||
}
|
||||
|
||||
/**
|
||||
* Gracefully drain all registered sessions on daemon shutdown. For each session that is
|
||||
* {@code BUSY}, poll up to {@code timeoutNanos} for it to leave {@code BUSY}, then release it
|
||||
* regardless. Non-busy sessions are released immediately. A failure releasing one session is
|
||||
* logged and does not abort the rest.
|
||||
* Gracefully drain all registered sessions on daemon shutdown. Non-busy sessions are released
|
||||
* immediately; a {@code BUSY} one is polled until it leaves {@code BUSY}, then released
|
||||
* regardless. A failure releasing one session is logged and does not abort the rest.
|
||||
*
|
||||
* <p>{@code timeoutNanos} is a budget for the WHOLE drain, not a grace period per session: the
|
||||
* deadline is taken once, before the loop. So the first BUSY session can spend all of it, and a
|
||||
* later BUSY one is then released with no wait at all. That is deliberate. This drain is only
|
||||
* one phase of shutdown — {@code Fleetd} closes the message service, the push loop, the
|
||||
* heartbeat, MCP and the router after it — and the whole sequence has to finish inside
|
||||
* launchd's exit window. A per-session grace would let N busy members drain for N * the
|
||||
* timeout, overrun that window, and get the daemon SIGKILLed part-way through; the members not
|
||||
* yet reached would then get no clean release, no preserved-worktree log, and no snapshot.
|
||||
* Cutting one turn short is the cheaper failure, and it is not silent: an abandoned BUSY
|
||||
* session is preserved, snapshotted, and logged at WARN by {@code logPreservedForShutdown}.
|
||||
*
|
||||
* <p>CB-544: this is a {@link ReleaseCause#SHUTDOWN} release — the worker's pane is stopped
|
||||
* (the process must end) but its worktree is preserved and its path logged. Shutdown is never
|
||||
|
||||
@@ -413,4 +413,29 @@ class CallerResolverTest {
|
||||
assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, null));
|
||||
assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, " "));
|
||||
}
|
||||
@Test
|
||||
void aWorkerOnAnyLoopbackSourceAddressIsStillAWorkerNotThePrimary() {
|
||||
// fleetd #305: the escalation. ConnectionIdentity used to accept only 127.0.0.1, so a
|
||||
// worker connecting from 127.0.0.2 resolved to no terminal, and this resolver's own
|
||||
// (wider) loopback check then made it the PRIMARY — granting spawn, stop, send and drain.
|
||||
// Measured on the Linux fleet host: binding a source of 127.0.0.2 succeeds there, so the
|
||||
// path is real and not theoretical.
|
||||
CallerResolver r = new CallerResolver(workerIdentity(), false, null);
|
||||
for (String src : new String[]{"127.0.0.1", "127.0.0.2", "127.1.2.3", "::ffff:127.0.0.2"}) {
|
||||
Principal p = r.resolve(src, 55555, null);
|
||||
assertEquals(Role.WORKER, p.role(), "a worker must stay a worker from source " + src);
|
||||
assertEquals("term_a", p.terminal(), "worker terminal from source " + src);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void aNonWorkerOnAnyLoopbackSourceAddressIsStillThePrimary() {
|
||||
// The other direction of the same fix: widening the identity check must not demote a
|
||||
// legitimate same-host primary that happens to connect from another 127.* address.
|
||||
CallerResolver r = new CallerResolver(nonWorkerIdentity(), false, null);
|
||||
for (String src : new String[]{"127.0.0.1", "127.0.0.2", "::ffff:127.0.0.1"}) {
|
||||
assertEquals(Role.PRIMARY, r.resolve(src, 55555, null).role(), "source " + src);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -40,6 +40,7 @@ public final class FakeHerdr implements HerdrClient {
|
||||
private final Map<String, List<String>> extraTabs = new LinkedHashMap<>();
|
||||
private int agentNameTakenFor = 0;
|
||||
private int agentPaneBusyFor = 0;
|
||||
private String agentStartErrorCode = null;
|
||||
private int workerTabPaneCount = 1;
|
||||
private String paneCloseErrorCode = null;
|
||||
private final Map<String, String> paneCloseErrorCodeFor = new ConcurrentHashMap<>();
|
||||
@@ -84,6 +85,12 @@ public final class FakeHerdr implements HerdrClient {
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make every {@code agent.start} call fail with this herdr error code. */
|
||||
public FakeHerdr agentStartFailsWith(String code) {
|
||||
this.agentStartErrorCode = code;
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Make the worker tab (w9:t2) report this many panes in {@code tab.list} (default 1). */
|
||||
public FakeHerdr withWorkerTabPaneCount(int n) {
|
||||
this.workerTabPaneCount = n;
|
||||
@@ -311,6 +318,10 @@ public final class FakeHerdr implements HerdrClient {
|
||||
+ required + "`", "invalid_request", null);
|
||||
}
|
||||
}
|
||||
if (agentStartErrorCode != null) {
|
||||
throw new HerdrException("herdr error [" + agentStartErrorCode + "]: agent.start failed",
|
||||
agentStartErrorCode, null);
|
||||
}
|
||||
long starts = calls.stream().filter(c -> c.method().equals("agent.start")).count();
|
||||
if (starts <= agentPaneBusyFor) {
|
||||
throw new HerdrException(
|
||||
|
||||
@@ -297,6 +297,68 @@ class InjectorTest {
|
||||
assertTrue(inj.activeTargets().isEmpty(), "the wedged target is reclaimed, not polled forever");
|
||||
}
|
||||
|
||||
/** A listener whose post-turn housekeeping always starts, as SessionManager's does with clearAfterTurn on. */
|
||||
private static final class PostTurnListener implements TurnListener {
|
||||
@Override public void onTurnComplete(String target) { }
|
||||
@Override public boolean hasPostTurnAction(String target) { return true; }
|
||||
@Override public boolean onTurnCompleteWithPostAction(String target) { return true; }
|
||||
}
|
||||
|
||||
@Test
|
||||
void aWorkerThatWedgesInUnknownAwaitingPostTurnPickupIsReleased() {
|
||||
// fleetd #306: the post-turn phase had no way out of a sustained unknown streak, so the
|
||||
// pickup latch stayed set, the target was polled forever, and every later message to it was
|
||||
// blocked by the delivery gate — while the session still looked healthy.
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver
|
||||
inj.onStatus(T, AgentStatus.WORKING); // turn starts
|
||||
inj.onStatus(T, AgentStatus.IDLE); // turn completes; housekeeping dispatched
|
||||
for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // then wedges
|
||||
|
||||
assertTrue(inj.activeTargets().isEmpty(),
|
||||
"a target wedged awaiting post-turn pickup must be reclaimed, not polled forever");
|
||||
assertEquals(List.of(), cap.failed,
|
||||
"the delegated turn already completed — a stuck /clear must not be reported as a failed turn");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aWorkerThatWedgesInUnknownAfterPickingUpTheResetIsReleased() {
|
||||
// The sibling latch. postTurnObserved is set when the reset is seen picked up (WORKING) and
|
||||
// is cleared only on a later injectable sample, so a wedge right after pickup sticks too.
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
inj.onStatus(T, AgentStatus.WORKING);
|
||||
inj.onStatus(T, AgentStatus.IDLE); // turn complete; reset dispatched
|
||||
inj.onStatus(T, AgentStatus.WORKING); // reset picked up -> postTurnObserved
|
||||
for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN);
|
||||
|
||||
assertTrue(inj.activeTargets().isEmpty(), "a wedge after reset pickup must also be reclaimed");
|
||||
assertEquals(List.of(), cap.failed, "still not a turn failure");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aBriefUnknownDuringPostTurnHousekeepingDoesNotDropTheLatch() {
|
||||
// The other direction: the escape must not fire on a glitch, or the queued next delegation
|
||||
// would overtake housekeeping that is still running.
|
||||
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
|
||||
inj.enqueue(T, "first", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "second", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
inj.onStatus(T, AgentStatus.WORKING);
|
||||
inj.onStatus(T, AgentStatus.IDLE); // first completes; reset dispatched
|
||||
for (int i = 0; i < 10; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // well under the grace
|
||||
|
||||
assertFalse(inj.activeTargets().isEmpty(), "a brief glitch must not release the post-turn latch");
|
||||
assertEquals(List.of("first"), sent(), "the queued delegation must not overtake housekeeping");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
|
||||
Captor cap = new Captor();
|
||||
|
||||
@@ -20,6 +20,17 @@ class ConnectionIdentityTest {
|
||||
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.0.0.1", 55555));
|
||||
}
|
||||
|
||||
@Test
|
||||
void resolvesWorkerFromAnyLoopbackSourceAddressNotJust127001() {
|
||||
// fleetd #305. On Linux the whole 127.0.0.0/8 is bound to lo, so a worker can connect with
|
||||
// a source address of 127.0.0.2. If identity resolution skips that address the caller has
|
||||
// no terminal, and a caller with no terminal is the primary under loopback-trust — so this
|
||||
// must resolve the worker, not null.
|
||||
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.0.0.2", 55555));
|
||||
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.1.2.3", 55555));
|
||||
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("::ffff:127.0.0.2", 55555));
|
||||
}
|
||||
|
||||
@Test
|
||||
void nullForOffHostCaller() {
|
||||
// A non-loopback peer can't be an on-host worker → treat as primary/unknown.
|
||||
|
||||
@@ -189,7 +189,7 @@ class FleetMcpTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
|
||||
void unansweredAsyncAskReturnsTheTicketToPendingThenAWorkersLateReplyStillCompletesIt() throws Exception {
|
||||
McpSchema.CallToolResult accepted = FleetMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
|
||||
String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
|
||||
|
||||
@@ -203,8 +203,15 @@ class FleetMcpTest {
|
||||
assertTrue(textOf(ask).contains("no answer"), textOf(ask));
|
||||
assertTrue(textOf(FleetMcp.poll(messages, ticket, null)).startsWith("[pending"));
|
||||
|
||||
// fleetd #307: the worker resumed on its own after the primary never answered, and its real
|
||||
// fleet_reply must complete its OWN async ticket — not strand in the inbox with
|
||||
// fleet_poll{ticket} stuck PENDING forever and later force-failed with a false "session
|
||||
// released before it replied" reason. This used to land in the inbox instead (see the old
|
||||
// assertion this replaced: messages.drainReplies("term_a").getFirst()...) — that was the bug.
|
||||
FleetMcp.reply(messages, "term_a", "finished after timeout");
|
||||
assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
|
||||
assertEquals("finished after timeout", textOf(FleetMcp.poll(messages, ticket, null)));
|
||||
assertTrue(messages.drainReplies("term_a").isEmpty(),
|
||||
"the reply completed its own ticket directly and never touched the inbox");
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -326,6 +333,26 @@ class FleetMcpTest {
|
||||
assertEquals("orphan", drained.getFirst().content());
|
||||
}
|
||||
|
||||
@Test
|
||||
void replyWithBlankContentIsACleanToolErrorNotAnUncaughtException() {
|
||||
// fleetd #302: MessageService.reply now REJECTS blank content by throwing. fleet_reply's
|
||||
// handler is a bare BiFunction with no try/catch around it, so if this guard only checked
|
||||
// `== null` (as it did), a whitespace-only reply would leave the handler as an uncaught
|
||||
// IllegalArgumentException instead of a tool error the caller can read. Null and whitespace
|
||||
// are the same caller mistake and must get the same answer — the sibling fleet_send guard
|
||||
// has always used isBlank for exactly this reason.
|
||||
for (String blank : new String[] {null, "", " ", "\n\t"}) {
|
||||
McpSchema.CallToolResult res = assertDoesNotThrow(
|
||||
() -> FleetMcp.reply(messages, "term_a", blank),
|
||||
"blank content must be refused as a tool error, never thrown out of the handler");
|
||||
assertEquals(Boolean.TRUE, res.isError(), "blank content is an error result");
|
||||
assertTrue(textOf(res).contains("content is required"),
|
||||
"the error names the missing argument: " + textOf(res));
|
||||
}
|
||||
assertEquals(0, messages.drainReplies("term_a").size(),
|
||||
"a refused reply must not reach the inbox");
|
||||
}
|
||||
|
||||
@Test
|
||||
void bridgePollWithTargetDrainsReplies() {
|
||||
// A reply with no open send queues it in the inbox.
|
||||
|
||||
@@ -1165,7 +1165,8 @@ class ClaudeCodeLauncherTest {
|
||||
@Test
|
||||
void spawnLetsAnUnrelatedHerdrErrorPropagateUnchanged() {
|
||||
// Fix 1 must only special-case a "*_not_found" answer. Any other herdr failure keeps
|
||||
// propagating as-is — this gate does not know how to recover from it.
|
||||
// propagating as-is — this gate does not know how to recover from it. The pane still needs
|
||||
// closing because spawn throws before it can return the pane id to a caller that could stop it.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown");
|
||||
herdr.agentGetFailsWithAfter(0, "internal_error");
|
||||
@@ -1182,8 +1183,27 @@ class ClaudeCodeLauncherTest {
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertEquals("internal_error", ex.code());
|
||||
assertEquals(0, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"an error this gate does not recognize is not this gate's teardown to run");
|
||||
assertEquals(1, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"the unchanged error leaves spawn without a pane id, so this gate closes its orphaned pane");
|
||||
}
|
||||
|
||||
@Test
|
||||
void panePlacementClosesTheSplitPaneWhenAgentStartFails() {
|
||||
FakeHerdr herdr = new FakeHerdr().agentStartFailsWith("internal_error");
|
||||
FleetConfig.Profile cfg = new FleetConfig.Profile(
|
||||
"ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
|
||||
List.of("claude"), "pane", "fleetd-workers", "w #{n}", null, null, null);
|
||||
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
|
||||
|
||||
dev.ltms.fleet.herdr.HerdrException ex = assertThrows(
|
||||
dev.ltms.fleet.herdr.HerdrException.class,
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertEquals("internal_error", ex.code(), "agent.start failure propagates unchanged");
|
||||
assertEquals(1, paneCloseCount(herdr, "w1:pSplit"),
|
||||
"the pane split for a peer that never starts is closed instead of left orphaned");
|
||||
}
|
||||
|
||||
// --- fleetd #176 fix 2: corroborated UNKNOWN refinement --------------------------------------
|
||||
|
||||
@@ -957,6 +957,76 @@ class MessageServiceTest {
|
||||
assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #307: a worker's {@code fleet_ask} can time out because the primary never answers —
|
||||
* distinct from {@link #aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket}, where the
|
||||
* primary DID answer and only its own bounded wait for the resumed turn expired.
|
||||
* {@code ask()}'s timeout path deliberately forgets the task's {@code turnId} (so
|
||||
* {@code hasAsyncQuestion} stops reporting the target BUSY — see
|
||||
* {@code unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget} above), which used
|
||||
* to also erase the one signal {@code askAnsweredAsyncTasks} needed to recognize the worker's
|
||||
* eventual real {@code fleet_reply}. That reply then had nowhere to land but the inbox, and
|
||||
* {@code fleet_poll{ticket}} stayed PENDING forever — later force-failed with the false reason
|
||||
* "session released before it replied", even though the worker had, in fact, replied.
|
||||
*/
|
||||
@Test
|
||||
void aReplyAfterAnAskTimeoutStillCompletesTheAsyncTicket() throws Exception {
|
||||
String ticket = messages.sendAsync(T, "task that asks then finishes alone");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
|
||||
assertEquals(MessageService.AskOutcome.TIMED_OUT,
|
||||
messages.ask(T, "which config?", 200).outcome());
|
||||
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
|
||||
"only the question wait ended; the delegated turn may still finish");
|
||||
|
||||
// The worker keeps working past the timeout and only now calls fleet_reply — with no live
|
||||
// rendezvous waiter open (ask()'s timeout already closed it) and no new send() having
|
||||
// reopened one for this target.
|
||||
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
|
||||
|
||||
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
|
||||
assertEquals("PR opened: https://example/pulls/42", done.reply(),
|
||||
"fleet_poll{ticket} must return the worker's real reply, not stay pending forever");
|
||||
assertEquals("reply", done.replySource());
|
||||
assertFalse(messages.hasStrandedReply(T),
|
||||
"the reply completed its own ticket directly and never touched the inbox");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
|
||||
* becomes false the instant it lapses — proven above), so a second, independent delegation can
|
||||
* be dispatched to the same target and itself go on to ask-and-lapse before the first worker's
|
||||
* real reply ever arrives. Two open tasks are then both eligible candidates on one target with
|
||||
* no live waiter to disambiguate them. A reply arriving now must not guess which one it answers
|
||||
* — guessing wrong would hand the lead a plausible-looking answer to a delegation the worker
|
||||
* never touched, worse than a failure because the lead acts on it — so it must fall back to the
|
||||
* inbox exactly as the zero-candidate case does.
|
||||
*/
|
||||
@Test
|
||||
void twoAskTimedOutTicketsOnOneTargetFallBackToTheInboxRatherThanGuess() throws Exception {
|
||||
String ticket1 = messages.sendAsync(T, "first task that asks");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q1?", 200).outcome());
|
||||
|
||||
String ticket2 = messages.sendAsync(T, "second task that asks");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q2?", 200).outcome());
|
||||
|
||||
assertTrue(messages.reply(T, "which task does this answer?"));
|
||||
|
||||
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket1).phase(),
|
||||
"an ambiguous reply must not guess ticket1");
|
||||
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket2).phase(),
|
||||
"an ambiguous reply must not guess ticket2");
|
||||
assertTrue(messages.hasStrandedReply(T));
|
||||
var drained = messages.drainReplies(T);
|
||||
assertEquals(1, drained.size());
|
||||
assertEquals("which task does this answer?", drained.get(0).content());
|
||||
}
|
||||
|
||||
@Test
|
||||
void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
|
||||
String first = messages.sendAsync(T, "first task");
|
||||
|
||||
@@ -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();
|
||||
@@ -327,7 +415,11 @@ class FleetAppTest {
|
||||
FakeHerdr herdr = new FakeHerdr().agentNameTakenTimes(99);
|
||||
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
|
||||
|
||||
assertEquals(500, req(port, "POST", "/members").statusCode());
|
||||
// fleetd #304: 502, not Javalin's default 500 — the herdr failure is named, and the body
|
||||
// carries herdr's own message, matching what fleet_spawn reports for the same failure.
|
||||
HttpResponse<String> res = req(port, "POST", "/members");
|
||||
assertEquals(502, res.statusCode());
|
||||
assertEquals("herdr_error", mapper.readTree(res.body()).get("error").asText());
|
||||
assertTrue(herdr.called("tab.create"), "a tab was created before the failed start");
|
||||
assertEquals("w9:t2", params(herdr, "tab.close").get("tab_id"), "orphaned tab must be closed");
|
||||
}
|
||||
@@ -444,6 +536,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();
|
||||
@@ -598,7 +737,12 @@ class FleetAppTest {
|
||||
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
|
||||
|
||||
// A genuine teardown failure must surface, not be reported as a successful 204.
|
||||
assertEquals(500, req(port, "DELETE", "/members/w9:pW").statusCode());
|
||||
// fleetd #304: it surfaces as a named 502 rather than Javalin's default 500. The property
|
||||
// this test guards is "not 204" and the herdr detail reaching the caller — a bare 500 gave
|
||||
// the body "Server Error" and said nothing about herdr.
|
||||
HttpResponse<String> res = req(port, "DELETE", "/members/w9:pW");
|
||||
assertEquals(502, res.statusCode());
|
||||
assertEquals("herdr_error", mapper.readTree(res.body()).get("error").asText());
|
||||
assertFalse(herdr.called("tab.close"), "tab is not removed when the pane close failed");
|
||||
}
|
||||
|
||||
@@ -611,4 +755,5 @@ class FleetAppTest {
|
||||
assertEquals(204, req(port, "DELETE", "/members/w9:pW").statusCode());
|
||||
assertTrue(herdr.called("tab.close"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user