Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 769f282408 | |||
| 1515025804 | |||
| 7754f53662 | |||
| a507f7b31b | |||
| 71c322f104 | |||
| fde2c15627 | |||
| 2830735644 | |||
| 4e3ac91a22 | |||
| 29d3f0b41f | |||
| cbc732444f | |||
| 6f828b8c38 | |||
| 145a8c8862 | |||
| 7057291739 | |||
| 2302b3bc11 | |||
| 3a004dc1b3 | |||
| 94ec77a1bc |
@@ -79,8 +79,11 @@ below are the procedure — run them in order, every task, not only the big ones
|
||||
the final judgment call, verification, merges, and anything that depends on context only you
|
||||
hold. Nothing else is yours by default.
|
||||
3. **Spawn every delegated unit first** — `fleet_spawn{profile, worktree:true, ticket}`, one per
|
||||
unit, *before* sending any. Pass `profile` explicitly: profiles differ in model and cost, not in
|
||||
tier, so the default is rarely what you want.
|
||||
unit, *before* sending any. Pass `profile` explicitly: profiles differ in model, cost and
|
||||
LIVENESS, not in tier, so the default is rarely what you want. The default is whatever the
|
||||
daemon reports, and on a host where it sits on an exhausted or withdrawn credential every
|
||||
unqualified spawn fails — sometimes loudly, sometimes as a member that spawns fine and then
|
||||
produces nothing. `fleet_profiles` reports the default; check it once per session.
|
||||
4. **Then send them all** — `fleet_send{sessionId, content, wait:false}`. Line 1 of every brief is
|
||||
`Load the <name> skill.` naming the worker's playbook; those skills are opt-in and that line is
|
||||
what makes them reliable. Where the project ships no such skill, spell the procedure out in the
|
||||
|
||||
@@ -192,7 +192,7 @@ Deliberately small; every one maps to a failure mode we have actually hit.
|
||||
| `fleet_send_duration_seconds` | histogram | delegated turn latency |
|
||||
| `fleet_replies_total{path}` | counter | path ∈ rendezvous\|inbox — how often a reply strands (CB-307's whole reason to exist) |
|
||||
| `fleet_inbox_depth{target}` | gauge | undrained replies; steady-state should be 0 |
|
||||
| `fleet_push_nudges_total{outcome}` | counter | outcome ∈ delivered\|exhausted — a rising `exhausted` means the primary is not draining |
|
||||
| `fleet_push_nudges_total{outcome}` | counter | outcome ∈ sent\|exhausted — a rising `exhausted` means the primary is not draining. `sent` was called `delivered` until fleetd #365; it counts the herdr paste-and-submit call returning, never a confirmation the pane read it |
|
||||
| `fleet_spawns_total{kind,outcome}` | counter | outcome ∈ ready\|timeout\|guard_rejected; per peer kind (CB-402) |
|
||||
| `fleet_sessions{state}` | gauge | SPAWNING/READY/BUSY/DONE census |
|
||||
| `fleet_herdr_calls_total{method,outcome}` | counter | socket health — the dependency everything rests on |
|
||||
|
||||
@@ -123,6 +123,11 @@ public final class Fleetd {
|
||||
// else can fail on a silently-empty one. A daemon started without a login shell (launchd)
|
||||
// boots fine either way — this is the only thing that says so out loud.
|
||||
reportRequiredSecrets(cfg);
|
||||
// fleetd #377: the git host value a member receives as GITEA_HOST is often a full URL
|
||||
// (scheme and trailing slash), not a host name. A member that assumes a bare host then
|
||||
// builds https://https://... and the request never leaves the machine. Report the shape
|
||||
// next to the secret report — shape only, never the value.
|
||||
reportGitHostShape(cfg);
|
||||
reportMemberTrustModel(cfg);
|
||||
// CB-596: an absent (or empty) memberCredentials: block blocks NOTHING — no credential
|
||||
// name is hardcoded any more to fall back on. Say so loudly, the same way a missing
|
||||
@@ -579,8 +584,12 @@ public final class Fleetd {
|
||||
if (cfg.health() != null && cfg.health().isEnabled()) {
|
||||
// CB-580: a member found GONE/NEVER_READY must fail whatever ticket is waiting on it,
|
||||
// through the same idempotent target-wide operation CB-516 already uses on release.
|
||||
// fleetd #386: System::nanoTime freezes across a macOS sleep, so the stall check also
|
||||
// gets a wall-clock source to detect and correct for that freeze. Every other decision
|
||||
// in FleetHealthMonitor stays on the monotonic clock, unchanged.
|
||||
healthMonitor = new FleetHealthMonitor(agents, sessions::roster, messages, healthScheduler,
|
||||
System::nanoTime, cfg.health().intervalOrDefault(),
|
||||
System::nanoTime, () -> TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis()),
|
||||
cfg.health().intervalOrDefault(),
|
||||
cfg.health().workingSuspectAfterOrDefault(), messages::abandon);
|
||||
String coverage = FleetHealthMonitor.coverage(true,
|
||||
cfg.health().notifications() != null && cfg.health().notifications().configured());
|
||||
@@ -1195,6 +1204,76 @@ public final class Fleetd {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #377: the env var names holding the git host value that members receive as
|
||||
* {@code GITEA_HOST}. {@code HerdrPeerLauncher.applyGitToken} injects {@code GITEA_HOST}
|
||||
* only for profiles that opted in via {@code gitTokenEnv} (CB-302), reading the value from
|
||||
* that profile's {@code gitHostEnv} (default {@code GITEA_HOST}), so the shape only matters
|
||||
* where a git token is opted in. A var used by more than one profile is one entry naming
|
||||
* every profile that reads it, the same shape as {@link #requiredSecretEnvVars}.
|
||||
*
|
||||
* <p>Package-private and pure (no I/O, no logging) so the derivation is unit-testable
|
||||
* without capturing log output; {@link #reportGitHostShape(FleetConfig)} is the logging caller.
|
||||
*/
|
||||
static Map<String, List<String>> gitHostEnvVars(FleetConfig cfg) {
|
||||
Map<String, List<String>> hostsBy = new LinkedHashMap<>();
|
||||
cfg.profiles().forEach((name, profile) -> {
|
||||
if (profile.hasGitToken()) {
|
||||
hostsBy.computeIfAbsent(profile.gitHostEnv(), _ -> new ArrayList<>())
|
||||
.add("profile '" + name + "' gitHostEnv");
|
||||
}
|
||||
});
|
||||
return hostsBy;
|
||||
}
|
||||
|
||||
/**
|
||||
* True when the value already starts with a URI scheme ({@code https://...}, {@code
|
||||
* http://...}). A bare host name and a host:port must both report {@code false} — the shape
|
||||
* this ticket exists for is a value that <em>looks like</em> a host but is a full URL, and
|
||||
* confusing those in the report would move the failure to the log instead of the network.
|
||||
*/
|
||||
static boolean startsWithScheme(String value) {
|
||||
return value.matches("[A-Za-z][A-Za-z0-9+.-]*://.*");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #377: log, on the same startup path as {@link #reportRequiredSecrets}, the SHAPE of
|
||||
* each git host value that members receive as {@code GITEA_HOST}: set or unset, its length,
|
||||
* whether it starts with a scheme, whether it ends with a slash. Never the value itself — the
|
||||
* same discipline as {@link #reportRequiredSecrets}, which logs by name only. Either shape is
|
||||
* legitimate: the value is passed to members unchanged, and a line that quietly rewrites it
|
||||
* would change what works on one host and breaks on another. The shape line only tells the
|
||||
* operator which URL form to expect from a member that builds on {@code GITEA_HOST}. An
|
||||
* unset variable is logged at INFO — useful information, not an error — and the daemon
|
||||
* starts on either way.
|
||||
*
|
||||
* <p>The env read lives in the overload below so a test can drive the line with a known
|
||||
* value and prove that value never reaches the log.
|
||||
*/
|
||||
static void reportGitHostShape(FleetConfig cfg) {
|
||||
reportGitHostShape(cfg, System.getenv());
|
||||
}
|
||||
|
||||
static void reportGitHostShape(FleetConfig cfg, Map<String, String> env) {
|
||||
Map<String, List<String>> hostsBy = gitHostEnvVars(cfg);
|
||||
if (hostsBy.isEmpty()) {
|
||||
log.info("startup git host: no profile sets a gitTokenEnv — nothing to check");
|
||||
return;
|
||||
}
|
||||
hostsBy.forEach((varName, sources) -> {
|
||||
String value = env.get(varName);
|
||||
if (value == null || value.isBlank()) {
|
||||
log.info("startup git host {}: unset ({}) — a member gets GITEA_TOKEN but no "
|
||||
+ "GITEA_HOST value", varName, String.join(", ", sources));
|
||||
} else {
|
||||
log.info("startup git host {}: set ({}) — length={}, startsWithScheme={}, "
|
||||
+ "trailingSlash={}",
|
||||
varName, String.join(", ", sources),
|
||||
value.length(), startsWithScheme(value), value.endsWith("/"));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #184: state the member trust model at startup. Environment controls and worktrees do
|
||||
* not make a sandbox when fleetd and its members use the same OS user. A separate herdr may
|
||||
|
||||
@@ -38,15 +38,46 @@ public final class FleetHealthMonitor {
|
||||
*/
|
||||
static final long ASK_LAPSE_RECHECK_DELAY_SECONDS = 120;
|
||||
|
||||
/**
|
||||
* fleetd #386: {@code System.nanoTime()} (or whatever {@link #clock} is) does not advance while
|
||||
* macOS sleeps, so a raw {@code nowNanos - lastActivityAtNanos} comparison freezes with the
|
||||
* host and can never cross {@link #workingSuspectAfterNanos}. This is a second, wall-clock
|
||||
* source used ONLY inside the stall check ({@link #stallElapsedNanos}) to detect and correct
|
||||
* for that freeze. Nothing else in this class reads it — every other decision (readiness grace,
|
||||
* the fault classification itself) stays exactly on {@link #clock}, as the ticket requires.
|
||||
*/
|
||||
private static final LongSupplier DEFAULT_REALTIME_CLOCK =
|
||||
() -> TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis());
|
||||
|
||||
private final AgentControl agents;
|
||||
private final Supplier<List<MemberSession>> roster;
|
||||
private final MessageService messages;
|
||||
private final ScheduledExecutorService scheduler;
|
||||
private final LongSupplier clock;
|
||||
private final LongSupplier realtimeClock;
|
||||
private final long intervalSeconds;
|
||||
private final long tickIntervalNanos;
|
||||
private final long workingSuspectAfterNanos;
|
||||
private final BiConsumer<String, String> failTarget;
|
||||
private final Map<String, HealthPrior> priors = new HashMap<>();
|
||||
/**
|
||||
* fleetd #386 clock-drift bookkeeping. {@code haveClockBaseline}/{@code lastTickMonoNanos}/
|
||||
* {@code lastTickRealNanos} track the previous tick's pair of readings so each new tick can
|
||||
* measure how far the two clocks moved apart since then. {@code accumulatedDriftNanos} is the
|
||||
* running total of every such divergence observed since this monitor started (never decreases —
|
||||
* the monotonic clock can only lag real time, never lead it). {@code busyDriftBaselineNanos}/
|
||||
* {@code busyBaselineActivityNanos} record, per target, the value of {@code accumulatedDriftNanos}
|
||||
* at the moment this monitor first saw that target's CURRENT {@code lastActivityAtNanos} while
|
||||
* BUSY — so {@link #stallElapsedNanos} adds back only the drift observed DURING this BUSY span,
|
||||
* never drift from a sleep that happened before the member went busy. All five fields are touched
|
||||
* only from {@code tick()}, like {@link #priors}.
|
||||
*/
|
||||
private boolean haveClockBaseline = false;
|
||||
private long lastTickMonoNanos;
|
||||
private long lastTickRealNanos;
|
||||
private long accumulatedDriftNanos = 0;
|
||||
private final Map<String, Long> busyDriftBaselineNanos = new HashMap<>();
|
||||
private final Map<String, Long> busyBaselineActivityNanos = new HashMap<>();
|
||||
/**
|
||||
* The live classification per member, and the only one of this class's three maps that more
|
||||
* than one scheduler task touches. {@code tick} writes it (and prunes it to the roster);
|
||||
@@ -88,12 +119,29 @@ public final class FleetHealthMonitor {
|
||||
public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
|
||||
ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds,
|
||||
long workingSuspectAfterSeconds, BiConsumer<String, String> failTarget) {
|
||||
this(agents, roster, messages, scheduler, clock, DEFAULT_REALTIME_CLOCK, intervalSeconds,
|
||||
workingSuspectAfterSeconds, failTarget);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param realtimeClock fleetd #386: a wall-clock nanosecond source (e.g.
|
||||
* {@code System.currentTimeMillis()} converted to nanos) that keeps
|
||||
* advancing while {@code clock} is frozen by a host sleep. Used only to
|
||||
* correct the stall check — see the class-level javadoc on the
|
||||
* clock-drift fields.
|
||||
*/
|
||||
public FleetHealthMonitor(AgentControl agents, Supplier<List<MemberSession>> roster, MessageService messages,
|
||||
ScheduledExecutorService scheduler, LongSupplier clock, LongSupplier realtimeClock,
|
||||
long intervalSeconds, long workingSuspectAfterSeconds,
|
||||
BiConsumer<String, String> failTarget) {
|
||||
this.agents = agents;
|
||||
this.roster = roster;
|
||||
this.messages = messages;
|
||||
this.scheduler = scheduler;
|
||||
this.clock = clock;
|
||||
this.realtimeClock = Objects.requireNonNull(realtimeClock, "realtimeClock");
|
||||
this.intervalSeconds = intervalSeconds;
|
||||
this.tickIntervalNanos = TimeUnit.SECONDS.toNanos(intervalSeconds);
|
||||
this.workingSuspectAfterNanos = TimeUnit.SECONDS.toNanos(workingSuspectAfterSeconds);
|
||||
this.failTarget = Objects.requireNonNull(failTarget, "failTarget");
|
||||
}
|
||||
@@ -123,6 +171,7 @@ public final class FleetHealthMonitor {
|
||||
for (Agent agent : agentsNow) live.put(agent.terminalId(), agent);
|
||||
HashSet<String> current = new HashSet<>();
|
||||
long nowNanos = clock.getAsLong();
|
||||
long driftBeforeThisTick = observeClockDrift(nowNanos);
|
||||
for (MemberSession session : rosterNow) {
|
||||
current.add(session.terminalId());
|
||||
Agent agent = live.get(session.terminalId());
|
||||
@@ -133,7 +182,7 @@ public final class FleetHealthMonitor {
|
||||
&& session.state() != MemberSession.State.SPAWNING;
|
||||
boolean readinessGraceElapsed = nowNanos - session.spawnedAtNanos() >= READINESS_GRACE_NANOS;
|
||||
boolean stalled = session.state() == MemberSession.State.BUSY
|
||||
&& nowNanos - session.lastActivityAtNanos() >= workingSuspectAfterNanos;
|
||||
&& stallElapsedNanos(session, nowNanos, driftBeforeThisTick) >= workingSuspectAfterNanos;
|
||||
// CB-643: the three message-layer facts CB-640 published. Read them here rather than
|
||||
// leaving them false — that constant is what made 8 of the 9 fault states dead.
|
||||
boolean queuedDelivery = messages.hasQueuedDelivery(session.terminalId());
|
||||
@@ -151,6 +200,8 @@ public final class FleetHealthMonitor {
|
||||
priors.keySet().retainAll(current);
|
||||
states.keySet().retainAll(current);
|
||||
orphanStreaks.keySet().retainAll(current);
|
||||
busyDriftBaselineNanos.keySet().retainAll(current);
|
||||
busyBaselineActivityNanos.keySet().retainAll(current);
|
||||
} catch (Throwable error) {
|
||||
// Any unclassified collection failure must never kill the monitor's only scheduler task.
|
||||
log.warn("fleet health collection failed; will retry next tick", error);
|
||||
@@ -175,6 +226,65 @@ public final class FleetHealthMonitor {
|
||||
return streak >= ORPHAN_CONFIRM_TICKS;
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #386: compare this tick's monotonic and real-time readings against the previous
|
||||
* tick's, and fold any positive divergence into {@link #accumulatedDriftNanos} (a ratchet — it
|
||||
* never decreases, since the monotonic clock can only fall behind real time, never ahead of
|
||||
* it). Logs once, at WARN, when that single tick's divergence exceeds one full tick interval —
|
||||
* the signature of a host that slept between the two ticks (a tick literally cannot run while
|
||||
* the process itself is suspended, so the whole sleep duration lands inside one tick's gap).
|
||||
*
|
||||
* @return {@link #accumulatedDriftNanos} as it stood BEFORE this tick's divergence was folded
|
||||
* in — the baseline {@link #stallElapsedNanos} needs when a target is observed BUSY
|
||||
* for the first time this tick, so a sleep that happened before this member went busy
|
||||
* is not attributed to it.
|
||||
*/
|
||||
private long observeClockDrift(long nowNanos) {
|
||||
long nowRealNanos = realtimeClock.getAsLong();
|
||||
long driftBeforeThisTick = accumulatedDriftNanos;
|
||||
if (haveClockBaseline) {
|
||||
long monoDelta = nowNanos - lastTickMonoNanos;
|
||||
long realDelta = nowRealNanos - lastTickRealNanos;
|
||||
long tickDrift = realDelta - monoDelta;
|
||||
if (tickDrift > tickIntervalNanos) {
|
||||
log.warn("fleet health: the monotonic clock did not advance for about {}s that the "
|
||||
+ "real clock did since the last tick (host likely slept); the stall "
|
||||
+ "detector could not see that time", TimeUnit.NANOSECONDS.toSeconds(tickDrift));
|
||||
}
|
||||
if (tickDrift > 0) {
|
||||
accumulatedDriftNanos = driftBeforeThisTick + tickDrift;
|
||||
}
|
||||
}
|
||||
lastTickMonoNanos = nowNanos;
|
||||
lastTickRealNanos = nowRealNanos;
|
||||
haveClockBaseline = true;
|
||||
return driftBeforeThisTick;
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #386: {@code nowNanos - lastActivityAtNanos} alone freezes across a host sleep, since
|
||||
* both come from the monotonic {@link #clock}. This adds back the real-time drift observed
|
||||
* since this BUSY span started — not the monitor's whole lifetime, so a sleep that happened
|
||||
* before this member went busy never leaks into its stall reading (see the class-level javadoc
|
||||
* on the drift fields). The baseline resets whenever {@code lastActivityAtNanos} changes (a new
|
||||
* turn) or the member is not currently BUSY.
|
||||
*/
|
||||
private long stallElapsedNanos(MemberSession session, long nowNanos, long driftBeforeThisTick) {
|
||||
String target = session.terminalId();
|
||||
if (session.state() != MemberSession.State.BUSY) {
|
||||
busyDriftBaselineNanos.remove(target);
|
||||
busyBaselineActivityNanos.remove(target);
|
||||
return nowNanos - session.lastActivityAtNanos();
|
||||
}
|
||||
Long baselineActivity = busyBaselineActivityNanos.get(target);
|
||||
if (baselineActivity == null || baselineActivity != session.lastActivityAtNanos()) {
|
||||
busyBaselineActivityNanos.put(target, session.lastActivityAtNanos());
|
||||
busyDriftBaselineNanos.put(target, driftBeforeThisTick);
|
||||
}
|
||||
long driftSinceBusyStart = accumulatedDriftNanos - busyDriftBaselineNanos.get(target);
|
||||
return (nowNanos - session.lastActivityAtNanos()) + driftSinceBusyStart;
|
||||
}
|
||||
|
||||
void reportTransition(String target, HealthState next) {
|
||||
HealthState previous = states.put(target, next);
|
||||
if (previous == next) return;
|
||||
|
||||
@@ -548,8 +548,19 @@ public final class CompletionResolver implements TurnListener {
|
||||
* fast completion. Runs the same backend-error classification the normal and raw-scrape paths
|
||||
* apply, against whatever is on screen right now: a match together with the too-fast crash
|
||||
* signature notifies {@link #backendErrorSink} (only on the resolution that wins the race). A
|
||||
* non-match stays the original generic too-fast failure, naming the member and both timings,
|
||||
* with whatever the pane shows appended so the caller sees the cause, not just "it failed".
|
||||
* non-match stays the generic too-fast failure, naming the member and both timings, with
|
||||
* whatever the pane shows appended so the caller sees the evidence, not just "it failed".
|
||||
*
|
||||
* <p>fleetd#376: <strong>this path must never resolve a completion.</strong> A fast backend can
|
||||
* genuinely answer inside the floor, so the failure is sometimes wrong — but it is wrong in the
|
||||
* loud direction, and the fix for that is honest wording, not a guess at the pane's meaning.
|
||||
* Reclassifying from the scrape was tried and rejected: there is no reliable positive marker for
|
||||
* "this is a real reply" across backends. {@link #lastAssistantBlock} falls back to the entire
|
||||
* pane when it finds no {@code ⏺} marker, so on a crash the candidate "reply" is the whole
|
||||
* screen; and {@code ⏺} itself is a Claude Code marker that an opencode pane never carries — the
|
||||
* very backend whose speed raised this ticket. Any weaker test (non-blank, or "contains sentence
|
||||
* punctuation") passes on almost every crash, because a pane holding a file path, a version
|
||||
* number or a hostname contains a full stop. That trades a loud wrong answer for a silent one.
|
||||
*/
|
||||
private void failTooFast(String target, InFlight turn, CompletableFuture<Rendezvous.Resolution> waiter,
|
||||
long elapsedNanos) {
|
||||
@@ -560,13 +571,13 @@ public final class CompletionResolver implements TurnListener {
|
||||
scrape = "";
|
||||
}
|
||||
String clippedScrape = clip(scrape);
|
||||
String baseReason = String.format(
|
||||
"member %s went BUSY -> DONE in %dms (floor %dms) — too fast to be real work, most "
|
||||
+ "likely a backend error before any work started",
|
||||
String timing = String.format(
|
||||
"member %s went BUSY -> DONE in %dms (floor %dms)",
|
||||
target, elapsedNanos / 1_000_000, MIN_TURN_NANOS / 1_000_000);
|
||||
String backendError = firstMatchingLine(scrape, backendErrorPatternOrFallback(target));
|
||||
if (backendError != null) {
|
||||
String reason = baseReason + ": " + clippedScrape;
|
||||
String reason = timing + " — too fast to be real work, and the pane carries a backend "
|
||||
+ "error: " + clippedScrape;
|
||||
if (rendezvous.resolveFailure(waiter, reason)) {
|
||||
inFlight.remove(target, turn);
|
||||
log.warn("failing send to {} via turn-stall fallback: {}", target, reason);
|
||||
@@ -575,7 +586,14 @@ public final class CompletionResolver implements TurnListener {
|
||||
}
|
||||
return;
|
||||
}
|
||||
fail(target, turn, clippedScrape.isBlank() ? baseReason : baseReason + ": " + clippedScrape);
|
||||
// fleetd#376: no pattern matched, so the cause is genuinely unknown. Say that, rather than
|
||||
// asserting a backend error the way this message used to — a fast backend really can finish
|
||||
// inside the floor, and a reader who trusts a wrong cause stops looking at the pane.
|
||||
String reason = timing + " — inside the floor. That is usually a backend error before any "
|
||||
+ "work started, but a fast backend can answer inside it too, and nothing here can "
|
||||
+ "tell those apart, so the turn is reported failed rather than guessed. Read the "
|
||||
+ "pane below before deciding which it was";
|
||||
fail(target, turn, clippedScrape.isBlank() ? reason : reason + ": " + clippedScrape);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -32,7 +32,11 @@ public interface LeadChannel {
|
||||
/** Non-destructive FIFO snapshot of the messages held for this daemon's own coord-id. */
|
||||
List<LeadMessage> peek();
|
||||
|
||||
/** Drop {@code msgId} from the held set and ack it on the broker. A no-op if it is not held. */
|
||||
/**
|
||||
* Drop {@code msgId} from the held set and ack it on the broker. A repeated ack that this
|
||||
* connection already completed may be a no-op. Any other unknown msgId must throw rather than
|
||||
* report an ack that did not reach the broker.
|
||||
*/
|
||||
void ack(String msgId);
|
||||
|
||||
/** This daemon's own lead coordination id — the mailbox it owns, and the {@code from} it sends as. */
|
||||
|
||||
@@ -5,6 +5,7 @@ import dev.ltms.fleet.herdr.AgentStatus;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
@@ -43,6 +44,9 @@ public final class LeadCoordLoop {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(LeadCoordLoop.class);
|
||||
|
||||
/** A bounded window is enough: redelivery can only follow a recent failed ack or recovery. */
|
||||
private static final int RECENT_DELIVERY_LIMIT = 1_024;
|
||||
|
||||
/** How an arriving peer message is rendered into the lead's pane — the sender's coord-id, then its text. */
|
||||
static final String DELIVERY_FORMAT = "[lead %s] %s";
|
||||
|
||||
@@ -51,6 +55,8 @@ public final class LeadCoordLoop {
|
||||
private final Supplier<Map<String, String>> leads;
|
||||
private final ScheduledExecutorService scheduler;
|
||||
private final long intervalMs;
|
||||
/** msgIds already written to the pane, so recovery redelivery is acked without another pane write. */
|
||||
private final LinkedHashMap<String, Boolean> delivered = new LinkedHashMap<>();
|
||||
|
||||
private volatile boolean running;
|
||||
|
||||
@@ -117,6 +123,13 @@ public final class LeadCoordLoop {
|
||||
if (held.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
LeadMessage msg = held.getFirst();
|
||||
if (wasDelivered(msg.msgId())) {
|
||||
// This lives here, rather than in LeadMailbox, because only this loop knows a pane write
|
||||
// happened. The mailbox only knows broker delivery tags and must still redeliver after a crash.
|
||||
ackDelivered(msg);
|
||||
return;
|
||||
}
|
||||
String lead = resolveLocalLead();
|
||||
if (lead == null) {
|
||||
// Left unacked on purpose: the broker keeps holding it until a lead pane exists.
|
||||
@@ -136,7 +149,6 @@ public final class LeadCoordLoop {
|
||||
lead, status, held.size());
|
||||
return;
|
||||
}
|
||||
LeadMessage msg = held.getFirst();
|
||||
try {
|
||||
agents.send(lead, DELIVERY_FORMAT.formatted(msg.from(), msg.content()));
|
||||
} catch (RuntimeException e) {
|
||||
@@ -145,6 +157,13 @@ public final class LeadCoordLoop {
|
||||
msg.msgId(), msg.from(), lead, e.toString());
|
||||
return;
|
||||
}
|
||||
rememberDelivered(msg.msgId());
|
||||
if (ackDelivered(msg)) {
|
||||
log.debug("lead coordination: delivered message {} from {} to lead {}", msg.msgId(), msg.from(), lead);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean ackDelivered(LeadMessage msg) {
|
||||
try {
|
||||
channel.ack(msg.msgId());
|
||||
} catch (RuntimeException e) {
|
||||
@@ -152,9 +171,24 @@ public final class LeadCoordLoop {
|
||||
// deliberate direction of this trade.
|
||||
log.warn("lead coordination: delivered message {} but could not ack it: {}",
|
||||
msg.msgId(), e.toString());
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private boolean wasDelivered(String msgId) {
|
||||
synchronized (delivered) {
|
||||
return delivered.containsKey(msgId);
|
||||
}
|
||||
}
|
||||
|
||||
private void rememberDelivered(String msgId) {
|
||||
synchronized (delivered) {
|
||||
delivered.put(msgId, Boolean.TRUE);
|
||||
if (delivered.size() > RECENT_DELIVERY_LIMIT) {
|
||||
delivered.remove(delivered.keySet().iterator().next());
|
||||
}
|
||||
}
|
||||
log.debug("lead coordination: delivered message {} from {} to lead {}", msg.msgId(), msg.from(), lead);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -59,7 +59,8 @@ import java.util.concurrent.TimeoutException;
|
||||
* <p><strong>Recovery.</strong> The connection is opened with automatic + topology recovery
|
||||
* enabled, mirroring {@code AmqpReplyInbox}: on reconnect the broker hands out fresh delivery tags,
|
||||
* so the held snapshot is cleared (dedup by {@code msgId} still prevents any double-queue on
|
||||
* redelivery) and any publish still awaiting its confirm is failed rather than left to idle out
|
||||
* redelivery). {@link LeadCoordLoop} separately deduplicates pane writes, since it alone knows
|
||||
* which messages reached a lead. Any publish still awaiting its confirm is failed rather than left to idle out
|
||||
* the confirm timeout against a sequence number that means nothing on the new channel.
|
||||
*/
|
||||
public final class LeadMailbox implements LeadChannel, AutoCloseable {
|
||||
@@ -85,6 +86,10 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
|
||||
private final Object channelLock = new Object();
|
||||
/** msgId → held delivery, for this mailbox's own queue only (there is exactly one). */
|
||||
private final LinkedHashMap<String, Held> held = new LinkedHashMap<>();
|
||||
/** Successful broker acks on this connection, retained only to make a repeated caller ack quiet. */
|
||||
private final LinkedHashMap<String, Boolean> recentlyAcked = new LinkedHashMap<>();
|
||||
/** Bounds {@link #recentlyAcked}: it is only an idempotency aid, never delivery state. */
|
||||
private static final int RECENT_ACK_LIMIT = 1_024;
|
||||
|
||||
/**
|
||||
* A dedicated channel for {@link #publish}, kept separate from {@link #channel} (consume + ack)
|
||||
@@ -162,16 +167,15 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
|
||||
}
|
||||
// On automatic recovery the broker redelivers unacked messages with FRESH delivery-tags; the
|
||||
// tags we were holding are now stale. Drop the held snapshot so the re-attached consumer
|
||||
// repopulates it with valid tags (dedup by msgId still prevents any double-queue). Any publish
|
||||
// repopulates it with valid tags (dedup by msgId still prevents any double-queue). LeadCoordLoop
|
||||
// remembers successful pane writes separately, so that redelivery cannot write a pane twice. Any publish
|
||||
// confirm still in flight when the connection dropped is equally stale — fail it now rather
|
||||
// than let it silently ride out CONFIRM_TIMEOUT_MS.
|
||||
if (connection instanceof Recoverable recoverable) {
|
||||
recoverable.addRecoveryListener(new RecoveryListener() {
|
||||
@Override
|
||||
public void handleRecovery(Recoverable recoverable) {
|
||||
synchronized (held) {
|
||||
held.clear();
|
||||
}
|
||||
clearHeldForRecovery();
|
||||
failPendingPublishesOnRecovery();
|
||||
log.info("AMQP lead mailbox connection recovered; cleared held messages for fresh redelivery");
|
||||
}
|
||||
@@ -350,15 +354,26 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
/** Remove the held message {@code msgId} and ack it on the broker. No-op if not held. */
|
||||
/**
|
||||
* Remove the held message {@code msgId} and ack it on the broker.
|
||||
*
|
||||
* <p>A repeated ack that this connection already completed is a no-op, tracked in the bounded
|
||||
* {@link #recentlyAcked} set. Any other missing entry throws: recovery clears {@link #held} while
|
||||
* the broker still owns the unacked delivery, and quiet success there would hide a required retry.
|
||||
* The set is bounded because it only distinguishes a recent duplicate caller ack from an unknown
|
||||
* delivery; it is not a substitute for broker state across a reconnect.
|
||||
*/
|
||||
@Override
|
||||
public void ack(String msgId) {
|
||||
Held h;
|
||||
synchronized (held) {
|
||||
h = held.remove(msgId);
|
||||
if (h == null && recentlyAcked.containsKey(msgId)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (h == null) {
|
||||
return; // never held (or already acked) — no-op
|
||||
throw new IllegalStateException("cannot ack lead message " + msgId + ": it is not held");
|
||||
}
|
||||
try {
|
||||
synchronized (channelLock) {
|
||||
@@ -372,6 +387,19 @@ public final class LeadMailbox implements LeadChannel, AutoCloseable {
|
||||
}
|
||||
throw new IllegalStateException("cannot ack lead message " + msgId, e);
|
||||
}
|
||||
synchronized (held) {
|
||||
recentlyAcked.put(msgId, Boolean.TRUE);
|
||||
if (recentlyAcked.size() > RECENT_ACK_LIMIT) {
|
||||
recentlyAcked.remove(recentlyAcked.keySet().iterator().next());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Clear stale delivery tags after recovery; package-private so the recovery contract test drives this exact path. */
|
||||
void clearHeldForRecovery() {
|
||||
synchronized (held) {
|
||||
held.clear();
|
||||
}
|
||||
}
|
||||
|
||||
private DeliverCallback deliverCallback() {
|
||||
|
||||
@@ -0,0 +1,241 @@
|
||||
package dev.ltms.fleet;
|
||||
|
||||
import ch.qos.logback.classic.Level;
|
||||
import ch.qos.logback.classic.Logger;
|
||||
import ch.qos.logback.classic.spi.ILoggingEvent;
|
||||
import ch.qos.logback.core.read.ListAppender;
|
||||
import dev.ltms.fleet.config.FleetConfig;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* fleetd #377: the startup git host line must report the <em>shape</em> of the value a
|
||||
* member receives as GITEA_HOST — set or unset, length, scheme, trailing slash — and the
|
||||
* value itself must never reach the log.
|
||||
*/
|
||||
class GitHostShapeReportTest {
|
||||
|
||||
/** A profile that opts in to the git-forge token, so GITEA_HOST is the var that matters. */
|
||||
private static final String GIT_TOKEN_CONFIG = """
|
||||
profiles:
|
||||
local:
|
||||
baseUrl: http://gx00.gw:8000
|
||||
gitTokenEnv: WORKER_GITEA_TOKEN
|
||||
""";
|
||||
|
||||
private static FleetConfig load(Path dir, String yaml) throws Exception {
|
||||
Path f = dir.resolve("fleetd.yaml");
|
||||
Files.writeString(f, yaml);
|
||||
return FleetConfig.load(f);
|
||||
}
|
||||
|
||||
/**
|
||||
* The level this logger had before {@link #attach()} raised it, so {@link #detach} can put it
|
||||
* back. {@code null} is a real value here — it means "inherit from the parent" — and that is
|
||||
* exactly the state this logger starts in, so it must be restored as {@code null} rather than
|
||||
* as some concrete level.
|
||||
*/
|
||||
private static Level originalLevel;
|
||||
|
||||
/**
|
||||
* fleetd #377: the shape lines are logged at INFO, and {@code logback-test.xml} sets
|
||||
* {@code dev.ltms.fleet} to WARN — so INFO events are dropped by the level check BEFORE any
|
||||
* appender sees them. Attaching an appender is therefore not enough: without raising the level
|
||||
* the list stays empty and every assertion below fails against correct production code. The
|
||||
* sibling report tests do the same thing at each call site (see
|
||||
* {@code MemberTrustModelReportTest}); doing it here keeps it in one place.
|
||||
*/
|
||||
private static ListAppender<ILoggingEvent> attach() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
|
||||
originalLevel = logger.getLevel();
|
||||
logger.setLevel(Level.INFO);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
return appender;
|
||||
}
|
||||
|
||||
private static void detach(ListAppender<ILoggingEvent> appender) {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(Fleetd.class);
|
||||
logger.detachAppender(appender);
|
||||
logger.setLevel(originalLevel);
|
||||
}
|
||||
|
||||
private static List<String> messages(ListAppender<ILoggingEvent> appender) {
|
||||
return appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
|
||||
}
|
||||
|
||||
@Test
|
||||
void setLineReportsLengthSchemeAndTrailingSlash(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
String value = "https://git.example.test/";
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of("GITEA_HOST", value));
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
String expected = "startup git host GITEA_HOST: set (profile 'local' gitHostEnv) — "
|
||||
+ "length=" + value.length() + ", startsWithScheme=true, trailingSlash=true";
|
||||
assertTrue(messages(appender).contains(expected),
|
||||
"a value with a scheme and a trailing slash must be reported by shape only: "
|
||||
+ "its length, startsWithScheme=true, trailingSlash=true");
|
||||
}
|
||||
|
||||
@Test
|
||||
void bareHostLineReportsNoSchemeNoTrailingSlash(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
String value = "git.example.test";
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of("GITEA_HOST", value));
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
String expected = "startup git host GITEA_HOST: set (profile 'local' gitHostEnv) — "
|
||||
+ "length=" + value.length() + ", startsWithScheme=false, trailingSlash=false";
|
||||
assertTrue(messages(appender).contains(expected),
|
||||
"a bare host name must report startsWithScheme=false, trailingSlash=false");
|
||||
}
|
||||
|
||||
@Test
|
||||
void unsetVariableIsLoggedAtInfoAndDoesNotThrow(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
// GITEA_HOST absent from the map entirely — the unset case must not throw.
|
||||
Fleetd.reportGitHostShape(cfg, Map.of());
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
assertTrue(appender.list.stream().anyMatch(e ->
|
||||
e.getLevel() == Level.INFO
|
||||
&& e.getFormattedMessage()
|
||||
.startsWith("startup git host GITEA_HOST: unset (profile 'local' gitHostEnv)")),
|
||||
"an unset git host is useful information, not an error — say it at INFO level");
|
||||
}
|
||||
|
||||
/**
|
||||
* The important test: the value must NEVER appear in the log output. This test fails if the
|
||||
* line is ever changed to include the value, because it drives the real logging path with a
|
||||
* value that carries a marker no shape field could contain.
|
||||
*/
|
||||
@Test
|
||||
void theValueNeverAppearsInLogOutput(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
String marker = "never-logged-host-shape-377";
|
||||
String value = "https://" + marker + "/";
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of("GITEA_HOST", value));
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
List<String> msgs = messages(appender);
|
||||
assertFalse(msgs.stream().anyMatch(m -> m.contains(value)),
|
||||
"the full GITEA_HOST value must never reach the log");
|
||||
assertFalse(msgs.stream().anyMatch(m -> m.contains(marker)),
|
||||
"no fragment of the value may reach the log — a line that embeds the value"
|
||||
+ " would leak at least this marker");
|
||||
assertTrue(msgs.stream().anyMatch(m -> m.contains("startsWithScheme=true")
|
||||
&& m.contains("trailingSlash=true")),
|
||||
"the shape must still be reported although the value is not");
|
||||
}
|
||||
|
||||
@Test
|
||||
void unsetReportAlsoCarriesNoValue(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
// A blank value is not injected either (putIfPresent skips it), so it must report unset
|
||||
// without echoing anything of it.
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of("GITEA_HOST", " "));
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
assertTrue(messages(appender).stream().anyMatch(m ->
|
||||
m.startsWith("startup git host GITEA_HOST: unset")),
|
||||
"a blank value is skipped by the launcher, so the line reports unset");
|
||||
}
|
||||
|
||||
@Test
|
||||
void defaultGitHostEnvIsGITEA_HOST(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, GIT_TOKEN_CONFIG);
|
||||
|
||||
assertEquals(Map.of("GITEA_HOST", List.of("profile 'local' gitHostEnv")),
|
||||
Fleetd.gitHostEnvVars(cfg));
|
||||
}
|
||||
|
||||
@Test
|
||||
void explicitGitHostEnvReportsUnderItsOwnName(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, """
|
||||
profiles:
|
||||
local:
|
||||
baseUrl: http://gx00.gw:8000
|
||||
gitTokenEnv: WORKER_GITEA_TOKEN
|
||||
gitHostEnv: MY_FORGE_HOST
|
||||
""");
|
||||
String value = "https://forge.example.test/";
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of("MY_FORGE_HOST", value));
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
assertTrue(messages(appender).contains(
|
||||
"startup git host MY_FORGE_HOST: set (profile 'local' gitHostEnv) — "
|
||||
+ "length=" + value.length()
|
||||
+ ", startsWithScheme=true, trailingSlash=true"),
|
||||
"an explicitly named gitHostEnv is reported under that name");
|
||||
}
|
||||
|
||||
@Test
|
||||
void noGitTokenMeansNoHostLineIsNeeded(@TempDir Path dir) throws Exception {
|
||||
FleetConfig cfg = load(dir, """
|
||||
profiles:
|
||||
local:
|
||||
baseUrl: http://gx00.gw:8000
|
||||
""");
|
||||
|
||||
ListAppender<ILoggingEvent> appender = attach();
|
||||
try {
|
||||
Fleetd.reportGitHostShape(cfg, Map.of());
|
||||
} finally {
|
||||
detach(appender);
|
||||
}
|
||||
|
||||
assertEquals(List.of("startup git host: no profile sets a gitTokenEnv — nothing to check"),
|
||||
messages(appender));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aHostWithAPortIsNotTreatedAsAScheme() {
|
||||
assertFalse(Fleetd.startsWithScheme("git.example.test"));
|
||||
assertFalse(Fleetd.startsWithScheme("git.example.test:3000"));
|
||||
assertFalse(Fleetd.startsWithScheme("git.example.test:3000/"));
|
||||
assertTrue(Fleetd.startsWithScheme("https://git.example.test"));
|
||||
assertTrue(Fleetd.startsWithScheme("https://git.example.test:3000/"));
|
||||
assertTrue(Fleetd.startsWithScheme("ssh://git.example.test"));
|
||||
}
|
||||
}
|
||||
+172
@@ -0,0 +1,172 @@
|
||||
package dev.ltms.fleet.config;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.RecordComponent;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
/**
|
||||
* Fleetd #358, the same "defect factory" #357 guarded on {@code FleetConfig.withDefaults()}
|
||||
* (see {@code FleetConfigWithDefaultsPreservesEveryComponentTest}), reproduced here on
|
||||
* {@link FleetConfig.Profile}. {@code Profile} carries a long back-compat constructor ladder — 8
|
||||
* constructors, re-counted directly against the source rather than trusted from the ticket, at
|
||||
* arities 25, 24, 22, 20, 18, 15, 14 and 12, against a canonical arity of 26 — and exactly ONE
|
||||
* rebuild site, {@link FleetConfig.Profile#withProfile(String)}, whose own
|
||||
* {@code return new Profile(...)} call is written at a literal 26-arg count. Add a 27th component
|
||||
* and its established back-compat constructor at the old (26-arg) arity, and {@code withProfile}'s
|
||||
* own call becomes a legal match for that new overload — silently dropping the new component every
|
||||
* time a profile's name is defaulted from its {@code workers:} key.
|
||||
*
|
||||
* <p>Builds one {@link FleetConfig.Profile} through the TRUE canonical constructor — resolved by
|
||||
* the record's own component types via {@code getDeclaredConstructor}, never by argument count —
|
||||
* with a real, distinctive, non-null value in every component, calls {@link
|
||||
* FleetConfig.Profile#withProfile(String)}, and asserts every component except {@code profile}
|
||||
* itself survives unchanged, while {@code profile} comes back as the new name it was given.
|
||||
*
|
||||
* <p>Every value here is chosen so {@code Profile}'s own compact constructor (which normalizes
|
||||
* several components — defaults {@code argv}/{@code kind}/{@code placement}/{@code workspace}/
|
||||
* {@code gitHostEnv}, nulls a handful of blank-checked strings, clamps {@code weight}, coerces
|
||||
* {@code subscription}) leaves it unchanged: every String is non-blank and already in the shape the
|
||||
* compact constructor would otherwise coerce it to (e.g. {@code placement} is already lowercase),
|
||||
* and every collection is non-empty. That is what makes "must survive unchanged" a valid assertion
|
||||
* for every component below, the same reasoning {@code FleetConfigWithDefaultsPreservesEveryComponentTest}
|
||||
* documents for {@code withDefaults()}.
|
||||
*
|
||||
* <p>{@link #EXCLUDED_FROM_SURVIVAL_CHECK} is kept deliberately empty and size-pinned by
|
||||
* {@link #exclusionListSizeIsPinned()} — a checker whose escape hatch can grow to silence a failure
|
||||
* is not a checker. Every one of {@code Profile}'s 26 current components has a real, non-null,
|
||||
* non-blank value here and none is excluded.
|
||||
*/
|
||||
class FleetConfigProfileWithProfilePreservesEveryComponentTest {
|
||||
|
||||
private static final RecordComponent[] COMPONENTS = FleetConfig.Profile.class.getRecordComponents();
|
||||
|
||||
/** Deliberately empty today; grow it only with a matching justification, and re-pin the size. */
|
||||
private static final Set<String> EXCLUDED_FROM_SURVIVAL_CHECK = Set.of();
|
||||
|
||||
/** One real, distinctive, non-null value per component, chosen to survive the compact ctor. */
|
||||
private static Map<String, Object> baseValues() {
|
||||
Map<String, Object> v = new LinkedHashMap<>();
|
||||
v.put("profile", "profile-guard");
|
||||
v.put("baseUrl", "https://guard.example/base");
|
||||
v.put("model", "model-guard");
|
||||
v.put("configDir", "/config/guard");
|
||||
v.put("tokenEnv", "GUARD_TOKEN");
|
||||
v.put("argv", List.of("guard-cmd"));
|
||||
v.put("placement", "guard-placement");
|
||||
v.put("workspace", "workspace-guard");
|
||||
v.put("tabLabel", "tab-guard");
|
||||
v.put("mcpUrl", "https://mcp.guard/");
|
||||
v.put("cwd", "/cwd/guard");
|
||||
v.put("parityOverlay", List.of(".guardrc"));
|
||||
v.put("gitTokenEnv", "GUARD_GIT_TOKEN");
|
||||
v.put("gitHostEnv", "GUARD_GIT_HOST");
|
||||
v.put("kind", "claude-code");
|
||||
v.put("env", Map.of("GUARD_ENV", "1"));
|
||||
v.put("weight", 2.5f);
|
||||
v.put("maxLoad", 4);
|
||||
v.put("subscription", Boolean.TRUE);
|
||||
v.put("exhaustedPattern", "pattern-guard");
|
||||
v.put("credentialId", "cred-guard");
|
||||
v.put("ideMcpUrl", "https://ide.guard/");
|
||||
v.put("ideProjectDir", "ide-project-guard");
|
||||
v.put("ideOpenCommand", "open-guard {dir}");
|
||||
v.put("autoCompactWindow", 150_000);
|
||||
v.put("errorPattern", "error-pattern-guard");
|
||||
assertNamesMatchComponents(v);
|
||||
return v;
|
||||
}
|
||||
|
||||
/**
|
||||
* Guards {@link #baseValues()} itself against drifting from the record's real shape — forgetting
|
||||
* to add a new component here fails this assertion by name, rather than silently checking one
|
||||
* component fewer than the record has.
|
||||
*/
|
||||
private static void assertNamesMatchComponents(Map<String, Object> values) {
|
||||
Set<String> names = new TreeSet<>();
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
names.add(rc.getName());
|
||||
}
|
||||
assertEquals(names, new TreeSet<>(values.keySet()),
|
||||
"this test's value map has drifted from FleetConfig.Profile's actual components — "
|
||||
+ "update baseValues() alongside the record");
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a {@link FleetConfig.Profile} through the TRUE canonical constructor — resolved by the
|
||||
* record's own component types, not by argument count — so this never accidentally exercises a
|
||||
* back-compat overload the way a literal {@code new Profile(...)} call risks doing.
|
||||
*/
|
||||
private static FleetConfig.Profile profileOf(Map<String, Object> values) throws ReflectiveOperationException {
|
||||
Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
|
||||
Object[] args = Arrays.stream(COMPONENTS).map(rc -> values.get(rc.getName())).toArray();
|
||||
Constructor<FleetConfig.Profile> ctor = FleetConfig.Profile.class.getDeclaredConstructor(types);
|
||||
return ctor.newInstance(args);
|
||||
}
|
||||
|
||||
@Test
|
||||
void exclusionListSizeIsPinned() {
|
||||
assertEquals(0, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
|
||||
"EXCLUDED_FROM_SURVIVAL_CHECK grew from 0 — every entry needs a justification in "
|
||||
+ "this test class's javadoc AND this assertion re-pinned to the new size; a "
|
||||
+ "growing exclusion list that silences failures on its own is not a guard");
|
||||
}
|
||||
|
||||
/**
|
||||
* The mutation this is built to catch: make {@code withProfile(String)}'s final constructor call
|
||||
* literal at some arg count, add one more component to the record with a new back-compat
|
||||
* constructor at the old arity, and the stale call silently rebinds. Every component here is real
|
||||
* and non-null/non-blank, so none of it should be replaced by {@code withProfile}, except
|
||||
* {@code profile} itself, which the method is documented to replace.
|
||||
*/
|
||||
@Test
|
||||
void withProfilePreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
Map<String, Object> base = baseValues();
|
||||
FleetConfig.Profile profile = profileOf(base);
|
||||
FleetConfig.Profile renamed = profile.withProfile("renamed-profile-guard");
|
||||
|
||||
List<String> dropped = new ArrayList<>();
|
||||
int checked = 0;
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
String name = rc.getName();
|
||||
if (EXCLUDED_FROM_SURVIVAL_CHECK.contains(name)) {
|
||||
continue;
|
||||
}
|
||||
checked++;
|
||||
Object expected = "profile".equals(name) ? "renamed-profile-guard" : base.get(name);
|
||||
Object actual;
|
||||
try {
|
||||
actual = rc.getAccessor().invoke(renamed);
|
||||
} catch (ReflectiveOperationException e) {
|
||||
throw new RuntimeException("failed to read FleetConfig.Profile." + name + "()", e);
|
||||
}
|
||||
if (!Objects.equals(expected, actual)) {
|
||||
dropped.add(String.format(Locale.ROOT,
|
||||
"%s: withProfile() was expected to carry (%s) for '%s' but returned %s — a "
|
||||
+ "component silently dropped by withProfile(), the shape of the "
|
||||
+ "defect this test exists to catch (its final \"return new "
|
||||
+ "Profile(...)\" call binding to a back-compat constructor instead "
|
||||
+ "of the true canonical one)",
|
||||
name, expected, name, actual));
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf(Locale.ROOT,
|
||||
"FleetConfig.Profile.withProfile() component-survival coverage — %d components, %d "
|
||||
+ "checked, %d excluded, %d survived%n",
|
||||
COMPONENTS.length, checked, EXCLUDED_FROM_SURVIVAL_CHECK.size(), checked - dropped.size());
|
||||
assertEquals(List.of(), dropped,
|
||||
"withProfile() silently dropped these components: " + dropped);
|
||||
}
|
||||
}
|
||||
@@ -206,6 +206,87 @@ class FleetHealthMonitorTest {
|
||||
.workingSuspectAfterOrDefault());
|
||||
}
|
||||
|
||||
// --- fleetd #386: a stall detector whose only clock freezes with a sleeping host is worse
|
||||
// than a silent one — it reports "quiet" for a member that was genuinely busy for hours.
|
||||
|
||||
@Test void monotonicClockFrozenPastThresholdOnRealClockStillReportsStallSuspected() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
try {
|
||||
AtomicLong mono = new AtomicLong(0);
|
||||
AtomicLong real = new AtomicLong(0);
|
||||
FakeHerdr herdr = new FakeHerdr().withAgent("busy", "term_busy", "pane_busy", "tab_busy");
|
||||
var scheduler = Executors.newSingleThreadScheduledExecutor();
|
||||
FleetHealthMonitor monitor = monitorWithClocks(herdr,
|
||||
List.of(member("term_busy", MemberSession.State.BUSY, 0, 0)), scheduler,
|
||||
mono::get, real::get, 60, 600, (_, _) -> { });
|
||||
|
||||
monitor.tick(); // establishes the clock baseline; nothing has diverged yet
|
||||
assertEquals(0, appender.list.stream().filter(event -> event.getFormattedMessage()
|
||||
.contains("state=STALL_SUSPECTED")).count());
|
||||
|
||||
// The host "sleeps": the monotonic clock stands completely still while the real clock
|
||||
// keeps moving, past the 600s stall threshold.
|
||||
real.set(TimeUnit.SECONDS.toNanos(700));
|
||||
monitor.tick();
|
||||
monitor.stop();
|
||||
|
||||
assertTrue(appender.list.stream().anyMatch(event -> event.getFormattedMessage()
|
||||
.contains("member=term_busy state=STALL_SUSPECTED")),
|
||||
"the real clock crossed the stall threshold even though the monotonic clock never moved");
|
||||
} finally {
|
||||
logger.detachAppender(appender);
|
||||
}
|
||||
}
|
||||
|
||||
@Test void clockDivergenceIsLoggedOnceNotOncePerTick() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
try {
|
||||
AtomicLong mono = new AtomicLong(0);
|
||||
AtomicLong real = new AtomicLong(0);
|
||||
var scheduler = Executors.newSingleThreadScheduledExecutor();
|
||||
FleetHealthMonitor monitor = monitorWithClocks(new FakeHerdr(), List.of(), scheduler,
|
||||
mono::get, real::get, 60, 600, (_, _) -> { });
|
||||
|
||||
monitor.tick(); // baseline: no divergence possible yet
|
||||
|
||||
// One sleep gap: the monotonic clock is frozen while the real clock jumps far past one
|
||||
// tick interval (60s).
|
||||
real.set(TimeUnit.SECONDS.toNanos(700));
|
||||
monitor.tick();
|
||||
|
||||
// The host is awake again: both clocks advance together from here, so no more divergence.
|
||||
mono.set(TimeUnit.SECONDS.toNanos(10));
|
||||
real.set(TimeUnit.SECONDS.toNanos(710));
|
||||
monitor.tick();
|
||||
mono.set(TimeUnit.SECONDS.toNanos(20));
|
||||
real.set(TimeUnit.SECONDS.toNanos(720));
|
||||
monitor.tick();
|
||||
monitor.stop();
|
||||
|
||||
assertEquals(1, appender.list.stream().filter(event -> event.getFormattedMessage()
|
||||
.contains("the monotonic clock did not advance"))
|
||||
.count(), "one sleep gap must produce exactly one divergence line, not one per tick");
|
||||
} finally {
|
||||
logger.detachAppender(appender);
|
||||
}
|
||||
}
|
||||
|
||||
private static FleetHealthMonitor monitorWithClocks(FakeHerdr herdr, List<MemberSession> roster,
|
||||
java.util.concurrent.ScheduledExecutorService scheduler, LongSupplier clock,
|
||||
LongSupplier realtimeClock, long intervalSeconds, long workingSuspectAfterSeconds,
|
||||
BiConsumer<String, String> failTarget) {
|
||||
AgentControl agents = new AgentControl(herdr);
|
||||
return new FleetHealthMonitor(agents, () -> roster,
|
||||
new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
|
||||
scheduler, clock, realtimeClock, intervalSeconds, workingSuspectAfterSeconds, failTarget);
|
||||
}
|
||||
|
||||
@Test void goneMemberRecoveryLogsOnceWithoutRefiringTargetFailure() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
|
||||
Level previousLevel = logger.getLevel();
|
||||
|
||||
@@ -357,6 +357,54 @@ class CompletionResolverTest {
|
||||
"the failure carries whatever was on screen: " + waiter.getNow(null).text());
|
||||
}
|
||||
|
||||
@Test
|
||||
void aPlausibleLookingReplyInsideTheFloorStillFails() {
|
||||
// fleetd#376 guard. A fix was attempted that inspected the pane inside the floor and resolved
|
||||
// a COMPLETION when the text "looked like a real reply". Every cheap test for that is unsafe:
|
||||
// lastAssistantBlock falls back to the WHOLE pane when there is no ⏺ marker, and a crash pane
|
||||
// almost always contains sentence punctuation — in a file path, a version, or a hostname.
|
||||
// This pane is the trap: it reads like a finished answer and it is a backend failure.
|
||||
FakeHerdr herdr = new FakeHerdr().readText(
|
||||
"Error: connection reset while loading src/main/java/Foo.java v1.2.3\n❯ ");
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
long[] clock = {10_000_000_000L};
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
|
||||
ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
var turn = new CompletionResolver.InFlight(waiter, null, clock[0]);
|
||||
clock[0] += CompletionResolver.MIN_TURN_NANOS - 1; // inside the floor
|
||||
|
||||
resolver.resolve("term_a", turn);
|
||||
|
||||
assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
|
||||
"inside the floor the verdict is always FAILED — never guess a completion from pane text");
|
||||
}
|
||||
|
||||
@Test
|
||||
void theTooFastFailureDoesNotAssertACauseItCannotKnow() {
|
||||
// fleetd#376: the message used to say "most likely a backend error before any work started".
|
||||
// When no error pattern matches, that cause is a guess, and a reader who believes it stops
|
||||
// looking at the pane. The verdict stays FAILED; only the claim about WHY is withdrawn.
|
||||
FakeHerdr herdr = new FakeHerdr().readText("I am running on opencode/mimo-v2.5-free.\n❯ ");
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
long[] clock = {10_000_000_000L};
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
|
||||
ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
var turn = new CompletionResolver.InFlight(waiter, null, clock[0]);
|
||||
clock[0] += CompletionResolver.MIN_TURN_NANOS - 1; // inside the floor
|
||||
|
||||
resolver.resolve("term_a", turn);
|
||||
|
||||
String text = waiter.getNow(null).text();
|
||||
assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(), "still fails, still loud");
|
||||
assertFalse(text.contains("most likely a backend error"),
|
||||
"an unmatched fast turn must not assert a backend error: " + text);
|
||||
assertTrue(text.contains("mimo-v2.5-free"), "the pane is still carried: " + text);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aBusyToDoneTransitionJustOutsideTheFloorResolvesNormally() {
|
||||
FakeHerdr herdr = new FakeHerdr().readText("⏺ a real, if quick, answer\n❯ ");
|
||||
@@ -1117,8 +1165,16 @@ class CompletionResolverTest {
|
||||
assertTrue(waiter.isDone());
|
||||
assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
|
||||
"still a failure — the floor itself, not the pattern, is why");
|
||||
assertTrue(waiter.getNow(null).text().contains("too fast to be real work"),
|
||||
"a non-match inside the floor stays the generic too-fast reason: " + waiter.getNow(null).text());
|
||||
// fleetd#376: this used to assert the phrase "too fast to be real work", which carried the
|
||||
// claim "most likely a backend error before any work started". With no pattern matched that
|
||||
// cause is a guess, so the wording was withdrawn. What this test really guards is unchanged:
|
||||
// the floor alone still fails the turn, it stays generic, and it never notifies the sink.
|
||||
String reason = waiter.getNow(null).text();
|
||||
assertTrue(reason.contains("inside the floor"),
|
||||
"a non-match inside the floor stays the generic floor reason: " + reason);
|
||||
assertFalse(reason.contains("most likely a backend error"),
|
||||
"a non-match must not assert a cause it did not establish: " + reason);
|
||||
assertTrue(reason.contains("still starting up"), "the pane is still carried: " + reason);
|
||||
assertTrue(notified.isEmpty(), "a non-match must never notify the typed sink");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,6 +54,40 @@ class LeadCoordLoopTest {
|
||||
assertTrue(channel.peek().isEmpty(), "and is no longer held");
|
||||
}
|
||||
|
||||
@Test
|
||||
void redeliveryOfAMessageAlreadyWrittenToThePaneIsAckedWithoutAnotherPaneWrite() {
|
||||
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "recover me"));
|
||||
var herdr = new FakeHerdr().agentStatus("idle");
|
||||
var loop = loop(channel, herdr, Map.of(LEAD_TERM, SELF));
|
||||
|
||||
loop.tick();
|
||||
channel.hold(new LeadMessage("m1", PEER, SELF, "recover me"));
|
||||
loop.tick();
|
||||
|
||||
assertEquals(1, prompts(herdr).size(), "a redelivery must not consume the lead pane twice");
|
||||
assertEquals(List.of("m1", "m1"), channel.acked(), "the redelivery still needs a fresh broker ack");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aRedeliveryIsAckedEvenWhileTheLeadIsMidTurn() {
|
||||
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "recover me"));
|
||||
var herdr = new FakeHerdr().agentStatus("idle");
|
||||
var loop = loop(channel, herdr, Map.of(LEAD_TERM, SELF));
|
||||
|
||||
loop.tick();
|
||||
channel.hold(new LeadMessage("m1", PEER, SELF, "recover me"));
|
||||
herdr.agentStatus("working");
|
||||
loop.tick();
|
||||
|
||||
assertEquals(1, prompts(herdr).size(), "the pane is still written exactly once");
|
||||
assertEquals(List.of("m1", "m1"), channel.acked(),
|
||||
"a message already written to the pane must be acked even mid-turn: the mid-turn "
|
||||
+ "gate exists to protect the pane, and this message needs no pane. Gating the ack "
|
||||
+ "on it leaves the message held on a lead that is busy most of the time, and every "
|
||||
+ "recovery redelivers it again — which is the loop this fix exists to stop");
|
||||
assertTrue(channel.peek().isEmpty(), "so it is no longer held");
|
||||
}
|
||||
|
||||
@Test
|
||||
void leavesTheMessageUnackedWhenTheLeadIsMidTurn() {
|
||||
var channel = new FakeLeadChannel(SELF).hold(new LeadMessage("m1", PEER, SELF, "hello"));
|
||||
|
||||
@@ -92,6 +92,33 @@ class LeadMailboxTest {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void ackThrowsWhenRecoveryClearedTheHeldMessage() throws Exception {
|
||||
String to = coordId("lead-recovery-ack");
|
||||
try (LeadMailbox inbox = LeadMailbox.open(uri(), to)) {
|
||||
inbox.publish(to, new LeadMessage("recovery-ack", "lead-from", to, "in flight"));
|
||||
assertEquals(1, awaitPeek(inbox).size(), "the broker delivery must be held before recovery clears it");
|
||||
|
||||
inbox.clearHeldForRecovery();
|
||||
|
||||
assertThrows(IllegalStateException.class, () -> inbox.ack("recovery-ack"),
|
||||
"a cleared delivery has no valid tag, so ack must report that it did not reach the broker");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void ackOfAMessageAlreadyAckedOnThisConnectionStaysQuiet() throws Exception {
|
||||
String to = coordId("lead-repeat-ack");
|
||||
try (LeadMailbox inbox = LeadMailbox.open(uri(), to)) {
|
||||
inbox.publish(to, new LeadMessage("repeat-ack", "lead-from", to, "once"));
|
||||
awaitPeek(inbox);
|
||||
|
||||
inbox.ack("repeat-ack");
|
||||
|
||||
inbox.ack("repeat-ack");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateMsgIdIsNotDoubleQueued() throws Exception {
|
||||
String to = coordId("lead-dedup");
|
||||
|
||||
@@ -1532,26 +1532,52 @@ class GitWorktreesTest {
|
||||
* for repo setup.
|
||||
*
|
||||
* <p>Scope, measured on the fleetd #369 merge and narrower than an earlier version of this
|
||||
* comment claimed: this protects the 5 {@link #seedingGitWorktrees} sites plus — through
|
||||
* comment claimed: this protects the {@link #seedingGitWorktrees} call sites plus — through
|
||||
* {@link #gitProcessBuilder} — every {@code git} subprocess the TEST itself starts. It does
|
||||
* NOT cover the other 53 {@code new GitWorktrees(...)} constructions in this file, which pass
|
||||
* NOT cover the {@code new GitWorktrees(...)} constructions elsewhere in this file that pass
|
||||
* no env override, so a production instance built that way still inherits the JVM's real
|
||||
* environment. Stripping this override from {@code seedingGitWorktrees} leaves the class green
|
||||
* both with and without the poison command above, so that half is currently unpinned.
|
||||
* environment. (Re-measured for fleetd #373, on this file as it stands here: 4 call sites go
|
||||
* through {@link #seedingGitWorktrees(Path, String, Path)} — not 5, an earlier count this
|
||||
* comment and fleetd #373's own ticket text both repeated without re-running it — out of 59
|
||||
* total {@code new GitWorktrees(...)} occurrences, one of which is the shared construction
|
||||
* inside {@link #seedingGitWorktrees(Path, String, Map)} itself. This class-wide count moves
|
||||
* every time a test is added, so treat any number here as a snapshot, not a fact to cite
|
||||
* without recounting.) Stripping the {@code gitEnv} override from a {@link
|
||||
* #seedingGitWorktrees} call site leaves the class green both with and without the poison
|
||||
* command above for that call site's OWN test, so that half was unpinned until fleetd #373
|
||||
* added {@link #seedingGitWorktreesResolvesTheExcludesFileFallbackInsideItsThrowawayDirectory}
|
||||
* below, which asserts the property directly instead of relying on a poisoned real machine.
|
||||
*/
|
||||
private static Map<String, String> hermeticGitEnv(Path tmp) {
|
||||
return hermeticGitEnvAt(tmp.resolve("hermetic-xdg-config-home-" + System.nanoTime()));
|
||||
}
|
||||
|
||||
/** Same isolation as {@link #hermeticGitEnv(Path)}, with an explicit {@code XDG_CONFIG_HOME}
|
||||
* instead of a fresh nanoTime-unique one under {@code tmp} — used by
|
||||
* {@link #seedingGitWorktreesResolvesTheExcludesFileFallbackInsideItsThrowawayDirectory}
|
||||
* (fleetd #373) so it can pre-populate that directory with a marker BEFORE the production
|
||||
* {@link GitWorktrees} instance reads it, something the random per-call name from
|
||||
* {@link #hermeticGitEnv(Path)} makes impossible to predict from outside. */
|
||||
private static Map<String, String> hermeticGitEnvAt(Path xdgConfigHome) {
|
||||
return Map.of(
|
||||
"GIT_CONFIG_GLOBAL", "/dev/null",
|
||||
"GIT_CONFIG_SYSTEM", "/dev/null",
|
||||
"GIT_TERMINAL_PROMPT", "0",
|
||||
"XDG_CONFIG_HOME", tmp.resolve("hermetic-xdg-config-home-" + System.nanoTime()).toString());
|
||||
"XDG_CONFIG_HOME", xdgConfigHome.toString());
|
||||
}
|
||||
|
||||
/** {@link GitWorktrees}'s full test seam, with a {@code memberSkillsSource} and no other
|
||||
* overrides — the shape every seeding test below needs, isolated via {@link #hermeticGitEnv}. */
|
||||
private static GitWorktrees seedingGitWorktrees(Path root, String memberSkillsSource, Path tmp) {
|
||||
return new GitWorktrees(root.toString(), null, _ -> {}, null, null, memberSkillsSource,
|
||||
hermeticGitEnv(tmp));
|
||||
return seedingGitWorktrees(root, memberSkillsSource, hermeticGitEnv(tmp));
|
||||
}
|
||||
|
||||
/** Same shape as {@link #seedingGitWorktrees(Path, String, Path)}, taking an already-built
|
||||
* {@code gitEnv} directly rather than computing one via {@link #hermeticGitEnv(Path)} — lets
|
||||
* fleetd #373's test drive the exact production construction a real member spawn uses, with a
|
||||
* {@code gitEnv} it has already pre-populated a marker into. */
|
||||
private static GitWorktrees seedingGitWorktrees(Path root, String memberSkillsSource, Map<String, String> gitEnv) {
|
||||
return new GitWorktrees(root.toString(), null, _ -> {}, null, null, memberSkillsSource, gitEnv);
|
||||
}
|
||||
|
||||
/** Acceptance criterion 2 (part 1): a worktree with no {@code .claude/} at all gets the skill
|
||||
@@ -1784,6 +1810,63 @@ class GitWorktreesTest {
|
||||
+ "after skill seeding ran — got:\n" + porcelain);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #373. Pins the production seam that fleetd #362 review finding 2 protects: {@link
|
||||
* GitWorktrees#previouslyEffectiveExcludesFileContent}'s XDG-fallback branch reads {@code
|
||||
* XDG_CONFIG_HOME}/{@code HOME} straight in Java, not through a {@code git} subprocess, so
|
||||
* {@code gitEnv} — the constructor seam every {@link #seedingGitWorktrees} instance in this
|
||||
* class is built with — is the ONLY thing that can isolate it. A mutation run during the
|
||||
* fleetd #372/#369 merge found this unpinned: replacing {@code hermeticGitEnv(tmp)} with
|
||||
* {@code null} in {@link #seedingGitWorktrees(Path, String, Path)} left every test in this
|
||||
* class green — the 56 tests that would fail against a real machine's poisoned {@code
|
||||
* XDG_CONFIG_HOME} were fixed by fleetd #369's subprocess-level isolation, but none of them
|
||||
* looks at what THIS Java-side read resolves, so deleting the override stays invisible.
|
||||
*
|
||||
* <p>This test asserts the PROPERTY, not the constructor argument: a {@link GitWorktrees}
|
||||
* built for seeding — through the very same {@link #seedingGitWorktrees(Path, String, Map)}
|
||||
* construction every other seeding test in this class goes through — must resolve the
|
||||
* excludes-file fallback inside its own throwaway {@code gitEnv}-supplied directory. It needs
|
||||
* NO externally-set poisoned environment variable: the marker pattern below is written ONLY
|
||||
* inside a throwaway {@code XDG_CONFIG_HOME} this test controls directly (bypassing {@link
|
||||
* #hermeticGitEnv(Path)}'s unpredictable nanoTime-named directory, via {@link
|
||||
* #hermeticGitEnvAt}, so the marker can be in place before the production instance ever reads
|
||||
* it), reachable ONLY through the {@code gitEnv} seam. If that seam is stripped, the
|
||||
* production code instead falls back to resolving the REAL {@code XDG_CONFIG_HOME}/{@code
|
||||
* HOME} of the machine running the test — which does not carry this marker — so the marker
|
||||
* file below shows up as untracked and the assertion fails on any machine, with no poison
|
||||
* command required. See the PR body for the pasted failure from actually running that
|
||||
* mutation (removing the {@code gitEnv} override from this test's own construction).
|
||||
*/
|
||||
@Test
|
||||
void seedingGitWorktreesResolvesTheExcludesFileFallbackInsideItsThrowawayDirectory(@TempDir Path tmp)
|
||||
throws Exception {
|
||||
Path xdgConfigHome = tmp.resolve("cb373-xdg-config-home");
|
||||
Files.createDirectories(xdgConfigHome.resolve("git"));
|
||||
Files.writeString(xdgConfigHome.resolve("git").resolve("ignore"), "cb373-xdg-fallback-marker\n");
|
||||
Map<String, String> gitEnv = hermeticGitEnvAt(xdgConfigHome);
|
||||
|
||||
Path repo = initRepo(tmp.resolve("repo"));
|
||||
Path skillsSource = tmp.resolve("skills-src");
|
||||
writeSkill(skillsSource, "implementer", "IMPLEMENTER SKILL\n");
|
||||
GitWorktrees seeding = seedingGitWorktrees(tmp.resolve("wts"), skillsSource.toString(), gitEnv);
|
||||
|
||||
String wt = seeding.add(repo.toString(), "cb-373-xdg-seam", "HEAD");
|
||||
assertEquals("IMPLEMENTER SKILL\n",
|
||||
Files.readString(Path.of(wt, ".claude", "skills", "implementer", "SKILL.md")),
|
||||
"fixture check — the skill really was seeded, so previouslyEffectiveExcludesFileContent ran");
|
||||
|
||||
Files.writeString(Path.of(wt, "cb373-xdg-fallback-marker"),
|
||||
"would only be invisible to git status if the fallback resolved THIS throwaway "
|
||||
+ "XDG_CONFIG_HOME rather than the real machine's\n");
|
||||
|
||||
String porcelain = fullStatus(Path.of(wt));
|
||||
assertEquals("", porcelain,
|
||||
"the marker pattern lives only in this test's throwaway XDG_CONFIG_HOME; git "
|
||||
+ "status must still be empty, proving the production seam resolved the "
|
||||
+ "excludes-file fallback through the gitEnv seam rather than the JVM's "
|
||||
+ "real environment — got:\n" + porcelain);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #369, acceptance criterion 4 — make the fix hard to undo by accident. Every git
|
||||
* subprocess this class starts is required to go through {@link #gitProcessBuilder}, the one
|
||||
|
||||
+190
@@ -0,0 +1,190 @@
|
||||
package dev.ltms.fleet.session;
|
||||
|
||||
import dev.ltms.fleet.peer.CharterReceipt;
|
||||
import dev.ltms.fleet.peer.MemberRole;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.RecordComponent;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
import java.util.function.Function;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
/**
|
||||
* Fleetd #358, the same "defect factory" #357 guarded on {@code FleetConfig.withDefaults()}
|
||||
* (see {@code FleetConfigWithDefaultsPreservesEveryComponentTest}), reproduced here on
|
||||
* {@link MemberSession} — the worse of the two sibling cases named in #358, because this record
|
||||
* has FIVE independent rebuild sites instead of one: {@link MemberSession#withState},
|
||||
* {@link MemberSession#withActivity}, {@link MemberSession#bumpTurn},
|
||||
* {@link MemberSession#withAgentSessionId} and {@link MemberSession#withFailureReason} each end in
|
||||
* their own literal {@code new MemberSession(...)} call. Add a 16th component and add the
|
||||
* established back-compat constructor at the old (15-arg) arity, and every one of those five
|
||||
* literal calls becomes a legal match for that new overload — silently dropping the new component,
|
||||
* independently, on whichever of the five paths a missed update leaves behind. That is harder to
|
||||
* spot than #357's single call site: the field would survive through some transitions and vanish
|
||||
* through others.
|
||||
*
|
||||
* <p>Each check below builds one {@link MemberSession} through the TRUE canonical constructor —
|
||||
* resolved by the record's own component types via {@code getDeclaredConstructor}, never by
|
||||
* argument count, so it can never itself land on a back-compat overload — with a real, distinctive,
|
||||
* non-null value in every component, calls the real rebuild method under test, and asserts every
|
||||
* component the method is not documented to change survives unchanged, while the component(s) it IS
|
||||
* documented to change come back as the new value it was given. A component that comes back
|
||||
* anything else was silently dropped or lost — the shape of the defect this test exists to catch.
|
||||
*
|
||||
* <p>{@link #EXCLUDED_FROM_SURVIVAL_CHECK} is kept deliberately empty and size-pinned by
|
||||
* {@link #exclusionListSizeIsPinned()}, for the same reason {@code FleetConfig}'s guard pins its own
|
||||
* exclusion list at zero: a checker whose escape hatch can grow to silence a failure is not a
|
||||
* checker. Every one of {@link MemberSession}'s 15 current components has a real, non-null,
|
||||
* non-blank value here and none is excluded.
|
||||
*/
|
||||
class MemberSessionRebuildPreservesEveryComponentTest {
|
||||
|
||||
private static final RecordComponent[] COMPONENTS = MemberSession.class.getRecordComponents();
|
||||
|
||||
/** Deliberately empty today; grow it only with a matching justification, and re-pin the size. */
|
||||
private static final Set<String> EXCLUDED_FROM_SURVIVAL_CHECK = Set.of();
|
||||
|
||||
/** One real, distinctive, non-null value per component — none of the 15 is excluded. */
|
||||
private static Map<String, Object> baseValues() {
|
||||
Map<String, Object> v = new LinkedHashMap<>();
|
||||
v.put("paneId", "pane-guard");
|
||||
v.put("terminalId", "term-guard");
|
||||
v.put("profile", "profile-guard");
|
||||
v.put("role", MemberRole.REVIEWER);
|
||||
v.put("cwd", "/wt/guard");
|
||||
v.put("ownerTerminal", "owner-guard");
|
||||
v.put("spawnedAtNanos", 111_111L);
|
||||
v.put("lastActivityAtNanos", 222_222L);
|
||||
v.put("turnCount", 7);
|
||||
v.put("state", MemberSession.State.BUSY);
|
||||
v.put("worktree", "/wt/guard-tree");
|
||||
v.put("branch", "worker/guard-branch");
|
||||
v.put("charterReceipt", new CharterReceipt(
|
||||
MemberRole.DEV, "profile-guard", "fleet.charters.dev", "deadbeefguard", 42));
|
||||
v.put("agentSessionId", "agent-guard");
|
||||
v.put("failureReason", "reason-guard");
|
||||
assertNamesMatchComponents(v);
|
||||
return v;
|
||||
}
|
||||
|
||||
/**
|
||||
* Guards {@link #baseValues()} itself against drifting from the record's real shape — forgetting
|
||||
* to add a new component here fails this assertion by name, rather than silently checking one
|
||||
* component fewer than the record has.
|
||||
*/
|
||||
private static void assertNamesMatchComponents(Map<String, Object> values) {
|
||||
Set<String> names = new TreeSet<>();
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
names.add(rc.getName());
|
||||
}
|
||||
assertEquals(names, new TreeSet<>(values.keySet()),
|
||||
"this test's value map has drifted from MemberSession's actual components — "
|
||||
+ "update baseValues() alongside the record");
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a {@link MemberSession} through the TRUE canonical constructor — resolved by the
|
||||
* record's own component types, not by argument count — so this never accidentally exercises a
|
||||
* back-compat overload the way a literal {@code new MemberSession(...)} call risks doing.
|
||||
*/
|
||||
private static MemberSession sessionOf(Map<String, Object> values) throws ReflectiveOperationException {
|
||||
Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
|
||||
Object[] args = Arrays.stream(COMPONENTS).map(rc -> values.get(rc.getName())).toArray();
|
||||
Constructor<MemberSession> ctor = MemberSession.class.getDeclaredConstructor(types);
|
||||
return ctor.newInstance(args);
|
||||
}
|
||||
|
||||
@Test
|
||||
void exclusionListSizeIsPinned() {
|
||||
assertEquals(0, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
|
||||
"EXCLUDED_FROM_SURVIVAL_CHECK grew from 0 — every entry needs a justification in "
|
||||
+ "this test class's javadoc AND this assertion re-pinned to the new size; a "
|
||||
+ "growing exclusion list that silences failures on its own is not a guard");
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared check for one rebuild site: build a base session with a real value in every component,
|
||||
* call {@code rebuild}, and assert every component comes back equal to {@code expectedOverrides}
|
||||
* when named there, or equal to the base value otherwise. Prints the same denominator style as
|
||||
* {@code FleetConfigWithDefaultsPreservesEveryComponentTest}.
|
||||
*/
|
||||
private void checkRebuildSite(String siteName, Function<MemberSession, MemberSession> rebuild,
|
||||
Map<String, Object> expectedOverrides) throws ReflectiveOperationException {
|
||||
Map<String, Object> base = baseValues();
|
||||
MemberSession session = sessionOf(base);
|
||||
MemberSession result = rebuild.apply(session);
|
||||
|
||||
List<String> dropped = new ArrayList<>();
|
||||
int checked = 0;
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
String name = rc.getName();
|
||||
if (EXCLUDED_FROM_SURVIVAL_CHECK.contains(name)) {
|
||||
continue;
|
||||
}
|
||||
checked++;
|
||||
Object expected = expectedOverrides.containsKey(name) ? expectedOverrides.get(name) : base.get(name);
|
||||
Object actual;
|
||||
try {
|
||||
actual = rc.getAccessor().invoke(result);
|
||||
} catch (ReflectiveOperationException e) {
|
||||
throw new RuntimeException("failed to read MemberSession." + name + "()", e);
|
||||
}
|
||||
if (!Objects.equals(expected, actual)) {
|
||||
dropped.add(String.format(Locale.ROOT,
|
||||
"%s: %s() was expected to carry (%s) for '%s' but returned %s — a component "
|
||||
+ "silently dropped by %s(), the shape of the defect this test exists "
|
||||
+ "to catch (its final \"return new MemberSession(...)\" call binding "
|
||||
+ "to a back-compat constructor instead of the true canonical one)",
|
||||
name, siteName, expected, name, actual, siteName));
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf(Locale.ROOT,
|
||||
"MemberSession.%s() component-survival coverage — %d components, %d checked, %d "
|
||||
+ "excluded, %d survived%n",
|
||||
siteName, COMPONENTS.length, checked, EXCLUDED_FROM_SURVIVAL_CHECK.size(),
|
||||
checked - dropped.size());
|
||||
assertEquals(List.of(), dropped,
|
||||
siteName + "() silently dropped these components: " + dropped);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withStatePreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
checkRebuildSite("withState", s -> s.withState(MemberSession.State.FAILED),
|
||||
Map.of("state", MemberSession.State.FAILED));
|
||||
}
|
||||
|
||||
@Test
|
||||
void withActivityPreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
checkRebuildSite("withActivity", s -> s.withActivity(999_999L),
|
||||
Map.of("lastActivityAtNanos", 999_999L));
|
||||
}
|
||||
|
||||
@Test
|
||||
void bumpTurnPreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
checkRebuildSite("bumpTurn", s -> s.bumpTurn(999_999L),
|
||||
Map.of("lastActivityAtNanos", 999_999L, "turnCount", 8));
|
||||
}
|
||||
|
||||
@Test
|
||||
void withAgentSessionIdPreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
checkRebuildSite("withAgentSessionId", s -> s.withAgentSessionId("agent-updated"),
|
||||
Map.of("agentSessionId", "agent-updated"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void withFailureReasonPreservesEveryOtherComponent() throws ReflectiveOperationException {
|
||||
checkRebuildSite("withFailureReason", s -> s.withFailureReason("reason-updated"),
|
||||
Map.of("failureReason", "reason-updated"));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user