Compare commits

...

38 Commits

Author SHA1 Message Date
Dai Ha ea12107497 fleetd #345: test timeout cancellation race
CI / contract (pull_request) Successful in 1m28s
CI / build (pull_request) Successful in 2m3s
2026-09-04 16:30:46 +07:00
Dai Ha 591df91de1 Merge #337: the deferred key set proves its own reporting coverage too
CI / contract (push) Successful in 54s
CI / build (push) Failing after 1m57s
2026-09-04 16:16:20 +07:00
Dai Ha 6a814176f0 #339: the start-of-line check must skip terminal chrome
CI / build (push) Failing after 1m30s
CI / contract (push) Successful in 1m54s
#339 stopped a member's own prose about an error from recording a credential
outage, by requiring the pattern at the start of its matched line. A bare
lookingAt also rejected a genuine error line rendered as

    | 503 Service Unavailable: upstream credential rejected

The send still failed, but the outage was never recorded. That is the false
negative #339's own invariant 3 named as worse than the false positive it set
out to fix: an unrecorded outage leaves the fleet spawning into a dead
credential.

Measured with a throwaway probe on the raw-scrape path, whose own comment says
to expect leading chrome there: kind=FAILED, sinkNotified=0.

startsWithBackendError now skips a leading run of non-letter, non-digit
characters before the check. That keeps #339's intent: prose still does not
match, because there the pattern sits after words rather than after chrome.
The worker's own prose test still passes.

Mutation: restoring the bare lookingAt fails the new test.
2026-09-04 16:13:19 +07:00
Dai Ha d11d1d157c Merge #339: a backend-error text match must be at the start of its line before it records a credential outage 2026-09-04 16:08:46 +07:00
Dai Ha d057d56156 #341: pin the noise control, and the reverse policy order
The fix reported every distinct unprotected name, but nothing held it there.

Mutation: replacing .filter(unprotectedGapNamesWarned::add) with a filter that
adds and always returns true - so every name is logged on every spawn - left
all 1358 tests green. The Set behaved; nothing proved this class used it as a
guard rather than as a record.

Two tests added:

- theSameUnprotectedNameIsWarnedAboutOnlyOnceAcrossSpawns pins invariant 1, the
  noise control. It now fails on that mutation, showing both duplicate WARNs.
- anAllowListWarnDoesNotSuppressALaterDenyByDefaultWarnForADifferentName covers
  the reverse policy order. The defect was found going deny-by-default then
  allow-list; a guard fixed in one direction is not fixed in the other.
2026-09-04 16:07:50 +07:00
Dai Ha d703ce1313 #337: extend ConfigRefTopLevelReportingCoverageTest to DEFERRED_KEYS
CI / contract (pull_request) Successful in 1m23s
CI / build (pull_request) Successful in 1m27s
ConfigRefTopLevelReportingCoverageTest (added by #333) proved every COLD_KEYS
and SPLIT_KEYS member has a real comparison behind it, but left
DEFERRED_TOP_LEVEL_KEYS unexercised. Re-measured by mutation (drop each
key's branch from changedDeferredKeys, run the suite, restore): 6 of the 11
deferred keys had no behavioural test naming them — guard, leadHeartbeat,
worktreeRoot, spawnReadyTimeoutMs, spawnReadyPollMs, quarantineCooldownSeconds
— which corrects the issue's own guessed list in two ways: lifecycle is
actually covered (ConfigRefTest.aDeferredChangeIsAppliedAndReported), and
worktreeRoot was missing from the issue's list entirely.

Promoted the test-side DEFERRED_TOP_LEVEL_KEYS copy into ConfigRef.DEFERRED_KEYS
(package-private, alongside COLD_KEYS/SPLIT_KEYS) so the reflective test reads
the same set changedDeferredKeys is compared against, and made
changedDeferredKeys package-private so the test can call it directly. Every
DEFERRED_KEYS component turned out to be a scalar or a simple record, so no
exclusion set was needed.

Mutation proof: dropping guard's branch from changedDeferredKeys leaves the
whole suite green except the new
everyDeferredKeyIsActuallyReportedByChangedDeferredKeys test, which fails
naming guard exactly.
2026-09-04 16:06:09 +07:00
Dai Ha b32a30fd47 Merge #341: warn once per distinct unprotected credential name, not once per launcher 2026-09-04 16:02:51 +07:00
Dai Ha 464dbc0930 fleetd#341: a per-name guard so a later spawn's different unprotected name still warns
CI / contract (pull_request) Successful in 1m1s
CI / build (pull_request) Successful in 1m29s
unprotectedGapLogged was one AtomicBoolean guarding two WARN branches in
logCredentialGap that name different env var names (the allow-list
keptByDerivedList branch, and warnGapUnprotected's deny-by-default /
non-zsh-fallback branch). memberCredentials is a live, re-read-per-spawn
supplier, so between two spawns a policy reload can change which names are
in the gap: spawn 1 warns about name A and trips the shared flag, and
spawn 2's gap containing a different name B never gets its WARN.

Replace the AtomicBoolean with unprotectedGapNamesWarned, a
ConcurrentHashMap-backed Set<String> guard keyed per name (same shape as
OpenCodeLauncher.modelCheckSkippedWarned), so each distinct credential-shaped
name is warned about exactly once, ever, regardless of which branch or
which spawn first reports it. allowListGapLogged (the separate INFO guard,
#192) is untouched. Neither WARN's wording changed.
2026-09-04 15:59:43 +07:00
Dai Ha a8cadd9150 Merge #338: a timed-out queued send cancels its message instead of leaving it to be delivered later
CI / contract (push) Successful in 1m23s
CI / build (push) Successful in 2m12s
2026-09-04 15:57:39 +07:00
Dai Ha 57b8c0b56d fleetd #339: guard backend error sink
CI / contract (pull_request) Successful in 50s
CI / build (pull_request) Successful in 1m50s
2026-09-04 15:57:18 +07:00
Dai Ha 147f50c19e #338: cancel timed-out queued deliveries
CI / contract (pull_request) Successful in 43s
CI / build (pull_request) Failing after 2m11s
2026-09-04 15:56:10 +07:00
Dai Ha eee4d576a2 #333: the Cold doc bullet listed four keys, COLD_KEYS has five
CI / build (push) Successful in 1m52s
CI / contract (push) Successful in 2m8s
memberHerdrSocket was missing from the prose. The #333 worker spotted it and
correctly left it alone as outside its scope.

Fixed by pointing the bullet at COLD_KEYS instead of re-listing its contents,
so the prose and the set cannot drift apart a second time.
2026-09-04 15:43:39 +07:00
Dai Ha b4f9d7f53a Merge #333: fleet: is a split key, and split membership now proves a reporting branch exists 2026-09-04 15:39:02 +07:00
Dai Ha 3aca53b967 fleetd#333: fleet.leaders is split too, and split membership now proves reporting exists
CI / contract (pull_request) Successful in 1m12s
CI / build (pull_request) Failing after 1m59s
F1: fleet: was sitting in ConfigRefTopLevelCoverageTest's HOT_EXCLUDED_TOP_LEVEL_KEYS
escape hatch, even though fleet.leaders is read only at startup (LeadTabScanner's
identity map, LeadLauncher.ensureLeads) while the rest of fleet: (role pools,
charters, tabLabel) is live. A reload changing only fleet.leaders reported a bare
"config reloaded" -- the operator edits a lead's tab: label, sees the reload
succeed, and the pane keeps resolving as a worker. Moved fleet into
ConfigRef.SPLIT_KEYS; changedSplitKeys now compares fleet.leaders specifically
(not the whole Fleet record, which would over-claim "restart" for a tabLabel-only
change) and names both halves in the message.

F2: membership in SPLIT_KEYS/COLD_KEYS never proved a matching branch existed in
changedSplitKeys/changedColdKeys -- measured by dropping the coordinator branch
while leaving "coordinator" in SPLIT_KEYS: both ConfigRefTopLevelCoverageTest and
the in-method "kept in step" assert stayed green. Added
ConfigRefTopLevelReportingCoverageTest, the ConfigRefProfileCoverageTest mechanism
one level up: reflection-built FleetConfig pairs that differ in exactly one
top-level component, calling the real (now package-private) changedColdKeys/
changedSplitKeys to prove each COLD_KEYS/SPLIT_KEYS member is actually reported.
Scoped to split+cold, not deferred -- see the new test's javadoc for why and what
that leaves open.

Both findings carry a behavioural test in ConfigRefTest plus a mutation proof
(revert -> real failure -> restore) recorded in the PR description.
2026-09-04 15:36:40 +07:00
Dai Ha 4aa1fae296 #329: a null task in answer() is not only "never an async ticket"
CI / contract (push) Successful in 59s
CI / build (push) Successful in 2m26s
The comment that landed with #329 said a null task means the turn was never
an async ticket. That is wrong, and it makes the guard read as complete.

A genuine async ticket also reaches answer() with task == null. ask() runs
clearAsyncQuestion(turnId, true) in its catch block, which drops the
asyncTasksByTurn entry, while rendezvous.closeAsk(turnId) runs later, in its
finally. Between the two the ask is still answerable and the map entry is
already gone, so answer()'s lookup returns null and the ticket is stranded.

Measured with a throwaway probe firing only that first half: answer() reported
REPLIED while the ticket stayed PENDING with a null reply. The probe used
forgetTurnForTest, so it omits markAskTimedOut; that cannot change the outcome,
because askTimedOut is read only by askAnsweredAsyncTasks, which reply() never
reaches while answer()'s own waiter is live.

Open as fleetd #334. The comment now says so.
2026-09-04 15:33:57 +07:00
Dai Ha 0c865032f9 Merge #329: log an exception thrown after a ticket resolves, complete the ticket from the task answer() already holds, and read orphan.turnId once 2026-09-04 15:26:45 +07:00
Dai Ha ea41bbf6b9 fleetd#329: fix silent async-ticket bugs in MessageService (F1/F2/F3)
CI / contract (pull_request) Successful in 50s
CI / build (pull_request) Failing after 1m28s
F2 (sendAsync executor catch): log when completeExceptionally returns
false, so an exception thrown after finishAsyncTask already completed
the ticket's future is no longer silently lost.

F1 (answer()'s stranded async ticket): reuse the Task reference answer()
already looked up before rendezvous.answerAsk(), instead of a second
asyncTasksByTurn lookup by turnId in finishAsyncTask. The second lookup
raced ask()'s unlocked timeout cleanup, which could forget turnId first
and leave the ticket stuck PENDING even though answer() itself returned
REPLIED. The #282 chained-ask guard is unaffected: it is still keyed on
result.outcome() == QUESTION, not on this lookup. Removed the now-unused
finishAsyncTask(String, Reply) overload.

F3 (reply()'s orphan recovery path): read orphan.turnId once instead of
twice, closing the same double-read shape fleetd #324 fixed in
finishAsyncTask.

Each fix has its own test plus a test-only race hook (mirroring #324's
finishAsyncTaskRaceHook) to force the exact interleaving deterministically.
Mutation-tested each fix by reverting it, confirming the real failure
(swallowed exception / PENDING ticket / NullPointerException), then
restoring it.

mvn clean install: Tests run: 1345, Failures: 0, Errors: 0, Skipped: 0,
BUILD SUCCESS.
2026-09-04 15:19:38 +07:00
Dai Ha 7b918c51ff Merge #330: a fourth reload class for split keys, and a top-level coverage checker
CI / build (push) Successful in 1m46s
CI / contract (push) Successful in 1m48s
2026-09-04 15:19:14 +07:00
Dai Ha 554395b104 fleetd#330: split reload class for health/coordinator + top-level coverage
CI / contract (pull_request) Successful in 53s
CI / build (pull_request) Successful in 2m17s
Unit 1: ConfigRef gets a fourth reload class, `split`, for keys read both
off the startup snapshot and live off config.get() at different sites
(health:, coordinator:). A split change is accepted (Outcome.applied()
stays true) and reported by name, naming which half is live and which
needs a restart, via a new Outcome.split() field kept separate from
deferred() since the two carry different guarantees for any caller that
branches on them, not just prose in summary(). Class doc updated: four
classes now, denominator note no longer calls health/coordinator
undecided.

Unit 2: ConfigRefTopLevelCoverageTest enumerates FleetConfig's 22
top-level record components and requires each to sit in exactly one of
COLD_KEYS, a pinned "compared in changedDeferredKeys" set, SPLIT_KEYS, or
a pinned hot-exclusion escape hatch — printing its own denominator and
pinning the escape hatch's exact contents the way #323 asked for.

Deviates from the issue's starting values by one key: `profiles` moves
from the suggested Hot bucket into the deferred bucket, because
changedDeferredKeys demonstrably compares it (add/remove and launch
settings), and citing "read live off the config supplier" for the whole
key would be false — most Profile fields are not read live, only
weight/maxLoad/credentialId are (and those are already covered by
ConfigRefProfileCoverageTest). Cold=5, split=2, deferred=11, hot=4,
total=22 — verified against the record and against ConfigRef's code, not
copied from the issue.
2026-09-04 15:16:45 +07:00
Dai Ha 823976c1b5 #326: state primary's real consequence, and write down the denominator
CI / contract (push) Successful in 56s
CI / build (push) Successful in 2m5s
The merged javadoc said a changed primary.terminal leaves a lead 'unresolved as
primary until a restart'. That over-claims. CB-532 made the pin deprecated:
identity comes from leaders:/leadScan:, and Fleetd.java:511 warns about the pin
at startup. A changed pin still needs a restart, but for the fallback nudge
destination, the deprecated identity path, and pushReminders/pushBackoffMs -
not for a lead that uses leaders:.

Also record what I measured. FleetConfig has 22 top-level components; four are
named nowhere in ConfigRef. memberCredentials and memberLoginShell are hot and
correctly absent (both read live off config.get() at spawn). health and
coordinator are undecided, not hot. 'Absent' looks the same for both kinds, and
twice now the forgotten kind hid among the correct kind.
2026-09-04 14:58:07 +07:00
Dai Ha c8388a7f92 Merge #326: primary and configReload are deferred keys, so a reload says a restart is needed 2026-09-04 14:53:50 +07:00
Dai Ha 02e6aef98c Merge #324: read task.turnId once in finishAsyncTask, so a concurrent clear cannot make the removal key null
CI / contract (push) Successful in 1m11s
CI / build (push) Successful in 1m29s
2026-09-04 14:44:42 +07:00
Dai Ha 6d493bc7bb fleetd#326: classify primary and configReload as deferred top-level keys
CI / contract (pull_request) Successful in 1m3s
CI / build (pull_request) Failing after 1m44s
ConfigRef.changedDeferredKeys only classified seven top-level FleetConfig
keys (#323 fixed the profile side). Two more keys are read only off the
startup snapshot and were missing:

- primary: Fleetd.java:506/519/520 feed PrimaryRegistry and ReplyPushLoop
  at construction; neither is rebuilt on reload.
- configReload: Fleetd.java:679-680 decide once at startup whether to
  build a ConfigWatcher at all, and with what interval; the watcher that
  would apply a later change is itself built once, so it is deferred
  (not cold — no already-open resource goes inconsistent, a running
  watcher just keeps its original settings).

health and coordinator are deliberately left unclassified: both are read
both off the startup snapshot AND live off the config supplier at a
second call site, so no single bucket is correct for either — see the
PR body for the options writeup and the coordinator.uriEnv exposure
question the issue asked to be answered.

Each fix is proven with a failing-first test in ConfigRefTest and a
revert-quote-restore mutation check (see PR body for the transcripts).
2026-09-04 14:42:24 +07:00
Dai Ha e5cb51a90e #324: read task.turnId once in finishAsyncTask to stop an NPE from ask()'s unlocked forgetting
CI / contract (pull_request) Successful in 1m28s
CI / build (pull_request) Successful in 2m1s
answer() holds sessionLocks while finishAsyncTask reads the volatile Task.turnId twice — once to
check it is non-null, once as the ConcurrentHashMap.remove key. ask()'s own timeout path mutates
the same field with no lock, via clearAsyncQuestion(turnId, true). volatile makes each read fresh
but not the pair atomic, so the field can go null between the two reads and remove(null, task)
throws NullPointerException on the lead's own answer() call, even though the reply already
completed on the line above.

Capture task.turnId into a local once and use that for both the check and the removal.

Added a package-private test seam (finishAsyncTaskRaceHook + forgetTurnForTest) so a test can force
the exact interleaving deterministically, by running the identical clearAsyncQuestion(turnId, true)
cleanup ask() uses, at the point between finishAsyncTask's former two reads. Both are inert (null)
in production.
2026-09-04 14:38:52 +07:00
Dai Ha e545c08082 #323: pin the exclusion set — the coverage mechanism's own escape hatch, found by mutation
CI / contract (push) Successful in 1m29s
CI / build (push) Successful in 2m10s
2026-09-04 14:32:08 +07:00
Dai Ha efa0deb9b2 Merge #323: the reload classifier proves its own coverage instead of claiming it 2026-09-04 14:29:10 +07:00
Dai Ha ca47e90c01 fleetd#323: close the reload-classifier drift with a reflection coverage test
CI / contract (pull_request) Successful in 1m11s
CI / build (pull_request) Successful in 2m6s
ConfigRef.sameLaunchSettings' javadoc claimed it compares every component
the launcher reads at spawn. It missed ideProjectDir, ideOpenCommand and
autoCompactWindow, and changedDeferredKeys separately missed worktreeGroup
(baked into the same GitWorktrees as worktreeRoot, Fleetd.java:251). A
reload that changed only one of those keys reported "config reloaded" with
nothing deferred, and the running daemon kept the old value.

Fix the four instances, and add ConfigRefProfileCoverageTest: it enumerates
every FleetConfig.Profile record component by reflection, mutates each one
not in the new ConfigRef.LAUNCH_SETTINGS_EXCLUDED set on a base profile,
and asserts sameLaunchSettings actually notices — so a fifth missed field
fails the build by name instead of drifting silently. It also prints its
own denominator (26 components, 23 compared, 3 excluded) per the ticket's
requirement that a checker must be able to state what it checked.

Also add the `profile` field itself to the comparison (it was neither
compared nor excluded before this fix — the coverage test surfaced it).

Rewrote the sameLaunchSettings javadoc to describe what the coverage test
actually guarantees instead of repeating the unchecked claim.
2026-09-04 14:26:09 +07:00
Dai Ha fa1f49675b Merge #318: a delivery landing after release is refused, not parked in a map nobody reads
CI / contract (push) Successful in 1m0s
CI / build (push) Successful in 2m8s
2026-09-04 14:21:29 +07:00
Dai Ha 8426c3528f #316: pin the fail-toward-preserve rule on the late re-check, found by mutation
CI / contract (push) Successful in 1m22s
CI / build (push) Successful in 1m42s
2026-09-04 14:20:36 +07:00
Dai Ha 65f98ba910 Merge #316: the dirty check that authorises the worktree removal is taken after the worker stops 2026-09-04 14:17:13 +07:00
Dai Ha d05205d1eb Add a hunter skill: a sweep and a diff review are different jobs with different output contracts
CI / contract (push) Successful in 47s
CI / build (push) Failing after 1m49s
2026-09-04 14:13:08 +07:00
Dai Ha 667254df47 #316: re-check worktree dirtiness after the pane stops, before removing it
CI / contract (pull_request) Successful in 39s
CI / build (pull_request) Successful in 1m46s
SessionManager.releaseRemoved read hasUncommitted() once, while the worker
could still write, then used that stale boolean after launcher.stop() to
authorise `git worktree remove --force`. The same stale read also gated
trySnapshot, so a worker that wrote between the read and the stop lost its
work with neither a preserve nor a snapshot.

Add a second, best-effort hasUncommitted read immediately before the
removal, taken only on the path that is actually about to delete something
(never on a release that already decided to preserve, and never for
SHUTDOWN, which preserves unconditionally). If the tree is now dirty,
preserve it and attempt a fresh snapshot, since the original snapshot never
ran when the pre-stop read said clean. A failing re-check also preserves,
matching the existing CB-581 fail-safe rule.
2026-09-04 14:12:49 +07:00
Dai Ha 2926cd1784 #318: release() no longer strands a delivery that lands while it is running
CI / build (pull_request) Failing after 1m59s
CI / contract (pull_request) Successful in 2m14s
AmqpReplyInbox.release() used held.remove(target) then iterated the old
map. A delivery landing on the consumer work-pool thread after the
remove (basicCancel does not flush one already handed to that pool) hit
deliverCallback's computeIfAbsent, found the key gone, and created a
brand-new map release() never looks at again — delivered-but-unacked
forever, never requeued, never redelivered (#298 only closed the
"already in held when release runs" case).

Fix: release() swaps in a RELEASED tombstone via held.compute(...)
instead of held.remove(...). ConcurrentHashMap serializes compute/
computeIfAbsent calls for the same key against each other, so whichever
of release() and a concurrent deliverCallback runs first is fully
visible to the other — no gap. deliverCallback checks for the
tombstone and nacks-with-requeue instead of recreating a map; peek/ack
treat it as empty; own() clears a stale tombstone so a target is never
poisoned if its id is ever reused (the issue's own text says id reuse
doesn't happen, but the tombstone would otherwise sit in `held` forever
either way).

New test AmqpReplyInboxReleaseRaceTest forces the actual interleaving
with a latch (blocks release() inside its nack loop, which is only
reachable after the tombstone swap, then fires a concurrent delivery)
rather than a sequential call — a sequential test would not have caught
this, since #298's own contract test forces settlement before release()
runs. Mutation-tested: reverting the fix makes this test fail with
"expected: <2> but was: <1>" (m1 never nacked); restored after
confirming that failure.
2026-09-04 14:10:21 +07:00
Dai Ha c801851c66 Correct the isLoopback javadoc: after #305 narrowing this range refuses a caller, it does not promote one
CI / contract (push) Successful in 1m16s
CI / build (push) Successful in 1m47s
2026-09-04 14:10:10 +07:00
Dai Ha de70aa38f1 Merge #317: an unresolved caller is refused, never promoted to primary
CI / contract (push) Successful in 1m1s
CI / build (push) Successful in 1m56s
2026-09-04 14:03:35 +07:00
Dai Ha 77ad88631b Merge #315: the fixed placement policy honours the retry loop's unreachable set
CI / build (push) Successful in 1m22s
CI / contract (push) Successful in 1m55s
2026-09-04 13:57:24 +07:00
Dai Ha d88017807b #315: fix self-contradicting javadoc left by the previous commit
CI / contract (pull_request) Successful in 1m21s
CI / build (pull_request) Successful in 2m31s
FixedPlacementPolicy's class javadoc still opened with "This ignores caps
and reachability" after the previous commit added reachability as the
fourth carve-out that is explicitly NOT ignored — caught by a shape-check
survey run against this same file as part of #315's own request ("look in
placement/ ... for the same shape: a caller/comment that documents an
expectation ... where an implementation does not meet it"). Reworded the
opening sentence: fixed still ignores caps (maxLoad) by design, but
reachability is now a narrower, per-call retry exclusion, not an ignored
concern.
2026-09-04 13:55:28 +07:00
Dai Ha 2159a5a94a #315: FixedPlacementPolicy now honors the retry loop's unreachable set
CI / contract (pull_request) Successful in 53s
CI / build (pull_request) Successful in 2m33s
CompositePeerLauncher.spawn retries a failed candidate on the next one and
rebuilds PlacementContext "so the policy excludes this profile" (its own
comment), but FixedPlacementPolicy.select never read ctx.unreachable(). Under
the default `fixed` placement policy (used when `placement` is unset or set
to `fixed`), every retry re-picked the same dead default and a second,
healthy, configured profile was never tried. This also covers the wiring-bug
branch (a candidate profile with no owning adapter), which hit the exact same
symptom for the same reason.

Not live on this fleet: fleetd.yaml sets placement: weighted, which already
consults ctx.unreachable() via PlacementPolicyUtil.available(). This is live
only for a deployment that leaves placement unset or sets it to fixed.

Fix is in FixedPlacementPolicy: consult ctx.unreachable() in the same two
places it already consults quarantined/coolingOff (the default check and the
fallback walk over candidates()), and add a fourth reason to the "no
candidate remains" exception. Considered fixing this in
CompositePeerLauncher's retry loop instead (break when select() returns an
already-unreachable profile), but that only fails faster on the same dead
profile — it cannot make the loop advance to a different candidate, because
only the policy decides which candidate is next. The defect is that one
policy implementation does not honor the loop's stated contract, so the fix
belongs in that policy, matching how weighted/round-robin already behave.

Also fixed: the "no reachable worker profile" exception message said
"trying N candidate(s)" where N was unreachable.size(), a count of DISTINCT
profiles (a HashSet dedupes a profile added twice), under wording that reads
as a count of attempts. Reworded to "N distinct candidate(s)" so the count
matches what is measured and the profile list that follows it.

Tests: two new failover tests next to the three existing ones in
CompositePeerLauncherTest (which all use PlacementPolicies.weighted(), which
is why this had no coverage) — one pinned to PlacementPolicies.fixed() for
the unreachable-default case, one for the wiring-bug (no adapter) case.
Mutation-proofed: reverted FixedPlacementPolicy.java, both new tests failed
with the exact bug ("no reachable worker profile available after trying 1
distinct candidate(s): a" / "...c"), then restored the fix.
2026-09-04 13:52:29 +07:00
26 changed files with 2965 additions and 134 deletions
+102
View File
@@ -0,0 +1,102 @@
---
name: hunter
description: Defect-hunt procedure for a fleetd worker — sweep an assigned package for real bugs and report several ranked findings without fixing anything. Load this when the lead asks you to hunt or audit a scope rather than review one diff. Do NOT load `reviewer` for this; the two want different output.
---
# Hunter worker — procedure
The turn contract (one `fleet_reply`, `fleet_ask` for the lead's decisions, honest reporting,
never merge) is in **`CLAUDE.md` → Bridge communication → Worker** and already applies.
**This skill is not `reviewer`.** `reviewer` judges one diff and reports the *single* most
important issue in about 90 words. A hunt sweeps a whole package and reports *several* findings
in a long structured form. Loading both gives you two contradictory output contracts, and the
usual result is a worker that writes a good report into its terminal and ends the turn without
sending it. Load exactly one.
## 0. Read this before you read code: how the report gets home
Your terminal reaches nobody. The lead sees **only** the text inside your `fleet_reply` call.
A long report is exactly the case where this goes wrong, so plan for it:
- **Write the report into the `fleet_reply` argument itself.** Do not compose it in your terminal
and then summarise it into the call.
- If the report is long, **send it anyway** — one `fleet_reply` with everything.
- If you end the turn without replying, the bridge scrapes your pane instead. That scrape carries
at most the last 4000 characters, and on a hunt it usually captures the tail of the lead's own
brief rather than your findings. The lead then has nothing and has to ask you again.
## 1. Change nothing
A hunt is read-only. Do not edit a production file, do not "quickly fix" what you find, and do
not run a formatter. You may run the build and tests to *check* a claim, and you should say so
when you did.
## 2. Read the whole scope first
Read every file in the assigned package before you judge any of it. A defect that a caller
elsewhere in the same package makes unreachable is not a defect, and you cannot know that from
one file.
Stay inside the scope. If a defect there depends on a class outside it, read that class to
confirm — but the defect itself must live in the scope you were given.
## 3. The bar — this matters more than the count
**Name the path into the bad state.** Say which caller, in which state, reaches it. A defect on
paper is not a reachable defect. If you cannot name that path, keep the finding but mark it
`unproven` and say exactly what you could not check. Do not drop it, and do not dress it up.
**Say which direction the harm goes.** Data loss, privilege escalation and silent wrong answers
are worth reporting even when the window is narrow. A finding whose worst outcome is a worse log
line is not worth a block.
Two workers once ran the same scope: the one that applied the direction-of-harm filter found ten
real defects, the one that did not found none. Fewer findings the lead can act on beat many the
lead has to triage.
## 4. Shapes that have produced real merged fixes here
Read for these first:
1. **A one-way gate.** A guard added after an incident closes only the direction that incident
came from. Do not only ask what closes the gate — ask **which states still open it**.
2. **A value read once, then used later to authorise something destructive**, after something
else has had a chance to change it.
3. **A failure downgraded to a value that looks like a legitimate result** — `-1`, `null`, an
empty list, `false` — which a caller then trusts.
4. **A lock held for one half of a read-modify-write and not the other**, or two collections
updated under different locks.
5. **A comment or javadoc stating an invariant the code no longer keeps.** Comments are
load-bearing in this repo; a stale one has already caused a bug.
## 5. What you cannot check, and must not claim you did
- `fleetd/fleetd.yaml` is gitignored and **absent from your worktree**. You cannot read it. If a
finding depends on live configuration, name the key and say you could not check it.
- `.mcp.json`, `opencode.json` and `.autoenv` in your worktree are neutralised stubs, not the
repo's real files.
- The `wiki/` submodule pointer is months old. Do not cite it.
Reporting a fact you took from the lead's brief as something you measured yourself is a false
report, even when the fact is correct. Say where each fact came from.
## 6. The report — what goes in `fleet_reply`
One block per finding, most severe first:
```
FINDING N — <one line>
file:line
Path in: <which caller, in which state, reaches this>
Direction: <data loss | escalation | silent wrong answer | outage | ...>
Window/trigger: <when it actually happens>
Confidence: <confirmed by reading | unproven — say what you could not check>
Why nothing else catches it: <the guard or test you checked, and why it misses>
```
End with one line naming every file you read, so the lead knows the denominator.
**Nothing clears the bar?** Reply `NO FINDINGS`, name the files you read, and say what you ruled
out. A clean sweep is a valid result; an invented defect is worse than none.
+5
View File
@@ -10,6 +10,11 @@ never merge) is in **`CLAUDE.md` → Bridge communication → Worker** and alrea
skill is only the *review procedure*: how to work the scope, and the exact shape of what you
send back.
**Wrong skill for a sweep.** This one reviews *one* diff or scope and reports the *single* most
important issue. If the lead asked you to hunt or audit a whole package for several defects, load
`hunter` instead and ignore this file — the two want different output, and following both is how a
worker ends its turn with a good report that never gets sent.
## 1. Read the whole scope before you judge
The delegation names your scope — a file, a diff, a PR, a function. **Read all of it first.**
+7 -2
View File
@@ -200,8 +200,13 @@ must obey belongs in the charter, not here.
adapter, with a message naming the credential and the remaining seconds ("cooling off after
repeated backend errors") — distinct wording from a quarantine refusal, so don't conflate the
two when reading a spawn failure.
- **Skills available to delegate:** `implementer` (worktree → commit → push → own PR) and
`reviewer` (scoped review → one structured finding). Name one in every delegation.
- **Skills available to delegate:** `implementer` (worktree → commit → push → own PR),
`reviewer` (one diff → one structured finding) and `hunter` (sweep a package → several ranked
findings, change nothing). Name exactly one in every delegation. **`reviewer` and `hunter` are
not interchangeable** — `reviewer` caps the answer at one finding in about 90 words, so naming
it for a multi-finding sweep hands the worker two contradictory output contracts. That has
already cost three workers' turns: each wrote a good report to its terminal and ended the turn
with no `fleet_reply`, and the scrape returned the tail of the brief instead.
- **Primary-side skills** (not delegation playbooks — a worker cannot use them):
`port-to-opencode` (make an OpenCode session a participant in this workspace) and
`fleets-status` (report every fleet that shares one LavinMQ instance).
@@ -22,46 +22,153 @@ import java.util.function.Supplier;
* choice rather than an accident of where the field was initialised.
*
* <h2>Not every key can change under a running daemon</h2>
* Keys fall into three classes, and the difference is about what already exists when the reload
* Keys fall into four classes, and the difference is about what already exists when the reload
* happens — not about how important the key is.
*
* <ul>
* <li><strong>Hot</strong> — re-read per use, so a reload takes effect on the next spawn:
* {@code fleet:} (every role pool, {@code charters}, and {@code tabLabel}),
* {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Those
* three are read through a supplier on {@code CompositePeerLauncher}, which is what makes
* them hot — not the fact that they are config. <strong>This does NOT include
* {@code fleet.leaders}</strong>: {@code Fleetd.main} reads {@code cfg.fleet().leaders()}
* once at startup to build the {@code LeadTabScanner} and the {@code LeadLauncher}, and
* neither is reconstructed on reload — so a lead added, removed, or re-{@code tab}'d under
* {@code fleet.leaders} needs a restart, the same as any deferred key below.</li>
* {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Both are
* read through a supplier on {@code CompositePeerLauncher}, which is what makes them hot —
* not the fact that they are config. Most of {@code fleet:} — every role pool
* ({@code architects}/{@code developers}/{@code reviewers}), {@code charters}, and
* {@code tabLabel} — is read the same live way, through the same supplier
* ({@code () -> config.get().fleet()}). <strong>But {@code fleet:} as a whole is NOT in this
* class</strong>: {@code fleet.leaders} inside the same key is frozen, which is exactly what
* makes {@code fleet:} split rather than hot — see below.</li>
* <li><strong>Deferred</strong> — accepted into the new snapshot, but the wiring built at startup
* keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:},
* {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code quarantineCooldownSeconds}
* (CB-578 stage B — baked once into the {@code BackendQuarantine} built at startup),
* {@code guard:}, {@code worktreeRoot:}, adding or removing a profile (a new backend needs its own launcher,
* {@code guard:}, {@code worktreeRoot:} and {@code worktreeGroup:} (both baked once into the
* {@code GitWorktrees} built at {@code Fleetd.java:251} and never rebuilt — fleetd #323
* instance 2 found {@code worktreeGroup} missing from this list and from
* {@link #changedDeferredKeys}), {@code primary:} (fleetd #326 — {@code Fleetd.java:506, 519,
* 520} read {@code cfg.primary()} only off the startup snapshot to build {@code
* PrimaryRegistry} and size {@code ReplyPushLoop}'s reminder cap/backoff, and neither is
* rebuilt on reload. Say the consequence exactly: {@code primary.terminal} is DEPRECATED
* (CB-532, and {@code Fleetd.java:511} warns about it at startup) — a lead's identity comes
* from {@code leaders:}/{@code leadScan:}, so changing this pin does not demote or promote a
* lead that uses those. What a changed pin still does not take effect on until a restart is
* the fallback nudge destination the pin remains, the deprecated identity path for an operator
* who still relies on it, and {@code pushReminders}/{@code pushBackoffMs}), {@code configReload:} (fleetd #326 — {@code
* Fleetd.java:679-680} read it only at startup to decide whether to build a {@code
* ConfigWatcher} at all and with what interval; the watcher that would apply a later change is
* itself built once, so a running watcher keeps polling on its original enabled flag and
* interval regardless of what a reload changes it to, the same shape as {@code lifecycle} —
* not cold, because no already-open resource goes inconsistent with the new value, the watcher
* (if any) simply keeps its old settings), adding or removing a profile (a new backend needs its own launcher,
* which is constructed once), <em>and an existing profile's launch settings</em> —
* {@code model}, {@code baseUrl}, {@code argv}, {@code env}, {@code mcpUrl},
* {@code exhaustedPattern} (CB-578 stage A — compiled once into {@code Fleetd.main}'s
* pattern map at startup), {@code errorPattern} (fleetd #201 Unit 5 — compiled once into
* {@code Fleetd.main}'s backend-error pattern map at startup, the same way), and the rest.
* {@code Fleetd.main}'s backend-error pattern map at startup, the same way),
* {@code ideProjectDir} / {@code ideOpenCommand} / {@code autoCompactWindow} (fleetd #323
* instance 1 — all three are read at spawn off the same frozen profile map and were missing
* from {@link #sameLaunchSettings}), and the rest of {@link #sameLaunchSettings}.
* {@code credentialId} (CB-578 stage B) is NOT on
* this list — it is read live off the config supplier at every quarantine check and
* exhaustion event, exactly like {@code weight} / {@code maxLoad}, so it is hot instead.
* {@code HerdrPeerLauncher} takes {@code Map.copyOf(profiles)} at construction and resolves
* each spawn out of that copy, so those never reach a launch until the daemon restarts. A
* reload logs these rather than pretending they applied.</li>
* <li><strong>Cold</strong> — cannot change at all under a running daemon: {@code bind:},
* {@code herdrSocket:}, {@code broker:} and {@code auth:}. The socket is bound, the broker
* <li><strong>Split</strong> (fleetd #330; extended to a third key by fleetd #333) — read
* <em>both</em> ways at different sites, so the key does not fit any class above as a whole:
* {@code health:}, {@code coordinator:} and {@code fleet:}. Each is read off the startup
* snapshot to build a long-lived object, and read live off {@link #get()} at a different,
* unrelated site — so half of a reload's effect already applies while the other half waits
* for a restart, and a bare "config reloaded" would under-claim by exactly that half.
* <ul>
* <li>{@code health:} — the monitor itself ({@code enabled}, {@code intervalSeconds},
* {@code workingSuspectAfterSeconds}) is built once at {@code Fleetd.java:556-563}
* and never rebuilt, so a changed value needs a restart to actually start, stop, or
* retime it. The coverage string {@code fleet_profiles} reports
* ({@code Fleetd.java:648-650}) is read live off {@link #get()} on every call, so it
* already reflects the new value.</li>
* <li>{@code coordinator:} — the {@code LeadMailbox} connection ({@code uri},
* {@code uriEnv}, {@code selfId}, {@code prefetch}) is opened once at
* {@code Fleetd.java:502} and never reopened, so a changed value needs a restart —
* {@code selfId} in particular names this daemon's own AMQP inbox queue, and a peer
* lead that learned the old name would not discover a new one on its own. The broker
* URI env-var <em>name</em> that {@code MemberEnvAllowList} keeps out of a member's
* environment is read live off {@link #get()} on every spawn
* ({@code HerdrPeerLauncher.java:1530}), so it already applies.</li>
* <li>{@code fleet:} (fleetd #333) — {@code fleet.leaders} is the frozen half:
* {@code Fleetd.java:281} reads {@code cfg.fleet().leaders()} off the startup snapshot
* for two long-lived objects built right after it and never rebuilt — the
* {@code LeadTabScanner}'s {@code tab label → lead name} map ({@code Fleetd.java:301},
* wired into {@code CallerResolver.withLeadsAndMembers} at {@code Fleetd.java:620/624},
* which is how a caller's pane is recognised as a lead at all) and, when herdr answered
* at startup, {@code LeadLauncher(...).ensureLeads()} ({@code Fleetd.java:315}), which
* auto-launches each declared lead up to its {@code instances} count. So a lead added,
* removed, or given a new {@code tab:} label under {@code fleet.leaders} needs a
* restart — until then it is invisible to identity resolution, and this is exactly the
* scenario fleetd #333 named: an operator edits a lead's {@code tab:} to match a
* renamed pane, sees "config reloaded", and the pane keeps resolving as a worker,
* because {@code CallerResolver} is still matching against the old label. The live
* half is the rest of {@code fleet:} — {@code architects}/{@code developers}/
* {@code reviewers}, {@code charters}, {@code tabLabel} — read live through the same
* {@code CompositePeerLauncher} supplier the Hot bullet above names, so a reload that
* only touches those already applies with nothing to report. Because the hot and frozen
* halves of {@code fleet:} are disjoint sub-fields rather than the same fields read two
* ways (contrast {@code coordinator.uriEnv} above), {@link #changedSplitKeys} compares
* {@code fleet.leaders} alone, not the whole {@code Fleet} record — comparing the whole
* record would report "split" for a {@code tabLabel}-only change that is actually fully
* hot, over-claiming in exactly the direction this class exists to avoid under-claiming
* in.</li>
* </ul>
* A split change is still accepted — {@link Outcome#applied()} stays {@code true}, the same
* as a deferred change — because the live half genuinely took effect; refusing the whole
* reload would leave the operator worse off than today. {@link Outcome#split()} names the
* key and says which half is which each time, rather than trying to score "how changed" a
* mixed key is or handle "both halves changed in one reload" as a special case.</li>
* <li><strong>Cold</strong> — cannot change at all under a running daemon. All five of
* {@link #COLD_KEYS}: {@code bind:}, {@code herdrSocket:}, {@code memberHerdrSocket:},
* {@code broker:} and {@code auth:}. The sockets are already connected, the broker
* connection is open, and the auth mode decides who may reach the port that is already
* listening.</li>
* listening. This bullet omitted {@code memberHerdrSocket:} until fleetd #333 — say "all
* five of COLD_KEYS" rather than re-listing them, so prose and set cannot drift again.</li>
* </ul>
*
* <p><strong>The denominator, measured on 2026-09-04 (fleetd #330; recounted for fleetd #333).</strong>
* {@code FleetConfig} has 22 top-level record components: 5 cold, 11 deferred, 3 split, 3
* hot-excluded. Three of them are named nowhere in this file, and the reason is the same for all
* three: {@code placement}, {@code memberCredentials} and {@code memberLoginShell} are
* <strong>hot</strong> and correctly absent — all three are read live off {@code config.get()}
* (placement through the {@code CompositePeerLauncher} supplier the Hot bullet names;
* {@code memberCredentials}/{@code memberLoginShell} at spawn time, {@code Fleetd.java:198, 205, 729}
* and {@code HerdrPeerLauncher#configuredMemberLoginShell}), so a reload takes effect on the next
* spawn with no entry needed here.
* {@code health} and {@code coordinator} used to be a third kind — <strong>undecided</strong>, not
* hot — until fleetd #330 added the <strong>split</strong> class above and gave them a home. A
* reload touching either used to report a bare "config reloaded", which under-claimed; now it names
* the key and says which half is which. {@code fleet} was the same story in reverse: fleetd #330's
* own fact-find named {@code health}/{@code coordinator} as "the complete set of split-shaped keys"
* and filed {@code fleet.leaders}'s restart requirement as a documented caveat sitting in the
* <strong>hot-excluded</strong> escape hatch instead — correctly documented, but in the one bucket
* this file's own coverage test cannot check the truth of (see that test's javadoc). fleetd #333
* moved it into <strong>split</strong>, where {@link #changedSplitKeys} actually reports it.
* <p>The point of writing the count down: "not mentioned in this file" looks identical for a key
* that is correctly hot and for a key nobody triaged. Three times now — {@code worktreeGroup} (#323),
* {@code primary}/{@code configReload} (#326), and {@code fleet.leaders} sitting in the escape hatch
* (#333) — the second kind hid among the first. A top-level coverage checker in the
* {@link ConfigRefProfileCoverageTest} shape (one level up, over {@code FleetConfig} itself rather
* than {@code FleetConfig.Profile}) proves this file's four classes exhaust the record's components
* — see {@code ConfigRefTopLevelCoverageTest}. That test proves the record's <em>shape</em> is fully
* triaged; it does NOT prove a {@code SPLIT_KEYS}/{@code COLD_KEYS}/{@code DEFERRED_KEYS} member has
* any reporting code behind it at all — {@code ConfigRefTopLevelReportingCoverageTest} is what
* fleetd #333 added for that, after measuring that a {@code SPLIT_KEYS} entry with its reporting
* branch deleted passes both this file's own "kept in step" assert and
* {@code ConfigRefTopLevelCoverageTest} unchanged. fleetd #337 extended it to {@code DEFERRED_KEYS}
* after measuring the same one-way gap there directly: dropping {@code guard}'s branch out of
* {@link #changedDeferredKeys} while {@code "guard"} stayed in the set left the whole suite green.
*
* <p><strong>A cold change refuses the whole reload.</strong> Not the hot half applied and the cold
* half warned about: that would leave the running daemon in a state matching no file on disk, which
* is the worst thing a reload can do to an operator debugging one. Refusing keeps the invariant that
* the live config is always some version of the file, and the message names the keys that must
* change through a restart.
* change through a restart. A split change does <em>not</em> refuse, for a different reason than a
* deferred change does not: its live half genuinely took effect, so refusing would throw that away
* and leave the operator worse off than the partial-but-honest report {@link Outcome#split()} gives.
*
* <p>A reload that fails to parse or fails validation is also refused, and the previous config keeps
* running. A config file being edited is normally read once mid-save; degrading a working daemon
@@ -71,10 +178,43 @@ public final class ConfigRef implements Supplier<FleetConfig> {
private static final Logger log = LoggerFactory.getLogger(ConfigRef.class);
/** Keys that cannot change under a running daemon — see the class doc. */
private static final Set<String> COLD_KEYS =
/**
* Keys that cannot change under a running daemon — see the class doc.
*
* <p>Package-private (not {@code private}) so {@code ConfigRefTopLevelCoverageTest} can fold it
* into the top-level triage it checks, the same way it reads {@link #SPLIT_KEYS}.
*/
static final Set<String> COLD_KEYS =
Set.of("bind", "herdrSocket", "memberHerdrSocket", "broker", "auth");
/**
* Keys read BOTH off the startup snapshot and live off {@link #get()} at different sites, so
* neither the hot, deferred nor cold class fits them as a whole — see the class doc's Split
* bullet (fleetd #330). A changed split key is accepted ({@link Outcome#applied()} stays
* {@code true}) and reported by name, with a message naming which half is live and which needs
* a restart.
*/
static final Set<String> SPLIT_KEYS = Set.of("health", "coordinator", "fleet");
/**
* Top-level keys {@link #changedDeferredKeys} compares — see the class doc's Deferred bullet.
* Promoted here from a test-side copy in {@code ConfigRefTopLevelCoverageTest} by fleetd #337,
* the same reason {@link #COLD_KEYS} and {@link #SPLIT_KEYS} live here rather than in a test: a
* second, hand-maintained copy of this set is exactly the kind of thing that silently drifts
* from the method it is supposed to describe. {@code spawnReadyTimeoutMs} and
* {@code spawnReadyPollMs} are compared together in one branch and reported under the combined
* label {@code "spawnReady*"}; {@code profiles} is compared twice over (added/removed names,
* then an existing profile's launch settings) — see {@link #changedDeferredKeys}.
*
* <p>Package-private (not {@code private}) so {@code ConfigRefTopLevelCoverageTest} and
* {@code ConfigRefTopLevelReportingCoverageTest} can both read it, the same way they already
* read {@link #COLD_KEYS} and {@link #SPLIT_KEYS}.
*/
static final Set<String> DEFERRED_KEYS = Set.of(
"guard", "worktreeRoot", "worktreeGroup", "primary", "configReload",
"leadHeartbeat", "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs",
"quarantineCooldownSeconds", "profiles");
private final Path path;
private final AtomicReference<FleetConfig> current;
@@ -102,25 +242,37 @@ public final class ConfigRef implements Supplier<FleetConfig> {
/**
* What a reload attempt did.
*
* <p>{@code split} is a separate field from {@code deferred} rather than a differently-worded
* entry inside it, because the two carry different guarantees for any caller that branches on
* them rather than just printing {@link #summary()}: every {@code deferred} entry means "this
* key's whole change waits for a restart", while every {@code split} entry means "part of this
* key's change already applied, and the message says which part" — collapsing them would force
* a caller to re-parse the message to tell those apart. See the class doc's Split bullet
* (fleetd #330) for why the key needs this at all.
*
* @param applied true when the new config is now live
* @param coldKeys cold keys whose value changed, which is why an unapplied reload was refused
* @param deferred keys that changed and were accepted, but whose effect waits for a restart
* @param deferred keys that changed and were accepted, but whose effect waits entirely on a
* restart
* @param split split keys that changed and were accepted, each named with which half of it
* is already live and which half waits for a restart
* @param error the parse or validation failure that refused the reload, else {@code null}
*/
public record Outcome(boolean applied, List<String> coldKeys, List<String> deferred,
String error) {
List<String> split, String error) {
public Outcome {
coldKeys = List.copyOf(coldKeys);
deferred = List.copyOf(deferred);
split = List.copyOf(split);
}
static Outcome refusedCold(List<String> keys) {
return new Outcome(false, keys, List.of(), null);
return new Outcome(false, keys, List.of(), List.of(), null);
}
static Outcome failed(String error) {
return new Outcome(false, List.of(), List.of(), error);
return new Outcome(false, List.of(), List.of(), List.of(), error);
}
/** A one-line summary for the operator — the reason, not just the verdict. */
@@ -132,11 +284,18 @@ public final class ConfigRef implements Supplier<FleetConfig> {
return "config reload refused — these keys cannot change under a running daemon: "
+ String.join(", ", coldKeys) + ". Restart fleetd to apply them.";
}
if (!deferred.isEmpty()) {
return "config reloaded; these changes need a restart to take effect: "
+ String.join(", ", deferred);
if (deferred.isEmpty() && split.isEmpty()) {
return "config reloaded";
}
return "config reloaded";
StringBuilder out = new StringBuilder("config reloaded");
if (!deferred.isEmpty()) {
out.append("; these changes need a restart to take effect: ")
.append(String.join(", ", deferred));
}
if (!split.isEmpty()) {
out.append("; partially live — ").append(String.join(" | ", split));
}
return out.toString();
}
}
@@ -177,14 +336,21 @@ public final class ConfigRef implements Supplier<FleetConfig> {
}
List<String> deferred = changedDeferredKeys(old, fresh);
List<String> split = changedSplitKeys(old, fresh);
current.set(fresh);
Outcome out = new Outcome(true, List.of(), deferred, null);
Outcome out = new Outcome(true, List.of(), deferred, split, null);
log.info(out.summary());
return out;
}
/** Cold keys whose value differs between the running config and the candidate. */
private static List<String> changedColdKeys(FleetConfig old, FleetConfig fresh) {
/**
* Cold keys whose value differs between the running config and the candidate.
*
* <p>Package-private (not {@code private}) so {@code ConfigRefTopLevelReportingCoverageTest}
* can call it directly with a reflection-built {@code FleetConfig} pair, the same reason
* {@link #sameLaunchSettings} is package-private — see that test's class doc (fleetd #333).
*/
static List<String> changedColdKeys(FleetConfig old, FleetConfig fresh) {
List<String> changed = new ArrayList<>();
if (!Objects.equals(old.bind(), fresh.bind())) {
changed.add("bind");
@@ -206,8 +372,16 @@ public final class ConfigRef implements Supplier<FleetConfig> {
return changed;
}
/** Changed keys that were accepted but whose effect waits for a restart. */
private static List<String> changedDeferredKeys(FleetConfig old, FleetConfig fresh) {
/**
* Changed keys that were accepted but whose effect waits for a restart.
*
* <p>Package-private (not {@code private}) so {@code ConfigRefTopLevelReportingCoverageTest}
* can call it directly with a reflection-built {@code FleetConfig} pair, the same reason
* {@link #changedColdKeys} and {@link #changedSplitKeys} already are (fleetd #333, extended to
* this method by fleetd #337 — membership in {@link #DEFERRED_KEYS} proved nothing about this
* method on its own until then; see that test's class doc).
*/
static List<String> changedDeferredKeys(FleetConfig old, FleetConfig fresh) {
List<String> changed = new ArrayList<>();
if (!Objects.equals(old.lifecycle(), fresh.lifecycle())) {
changed.add("lifecycle");
@@ -221,6 +395,29 @@ public final class ConfigRef implements Supplier<FleetConfig> {
if (!Objects.equals(old.worktreeRoot(), fresh.worktreeRoot())) {
changed.add("worktreeRoot");
}
// Baked into the same GitWorktrees as worktreeRoot (Fleetd.java:251) and never rebuilt
// either — see the class doc. Missing this check was fleetd #323 instance 2: a reload
// that changed only worktreeGroup reported "config reloaded" with nothing deferred, and
// newly provisioned worktrees kept the old sharing behaviour.
if (!Objects.equals(old.worktreeGroup(), fresh.worktreeGroup())) {
changed.add("worktreeGroup");
}
// fleetd #326: Fleetd.java:506, 519, 520 read cfg.primary() only off the startup snapshot
// (PrimaryRegistry's pinned terminal, ReplyPushLoop's reminder cap and backoff) — neither is
// rebuilt on reload, so a changed value needs a restart. Note what it does NOT mean:
// primary.terminal is deprecated (CB-532), identity comes from leaders:/leadScan:, so a lead
// using those is unaffected by this pin either way. See the class doc for the exact scope.
if (!Objects.equals(old.primary(), fresh.primary())) {
changed.add("primary");
}
// fleetd #326: Fleetd.java:679-680 read cfg.configReload() only at startup to decide whether
// to build a ConfigWatcher at all and with what interval — the watcher that would apply a
// later change is itself built once, so a running watcher keeps its original enabled flag and
// interval regardless of what a reload changes it to. Not cold: no already-open resource goes
// inconsistent with the new value, a watcher (if any) simply keeps polling on the old settings.
if (!Objects.equals(old.configReload(), fresh.configReload())) {
changed.add("configReload");
}
if (!Objects.equals(old.spawnReadyTimeoutMs(), fresh.spawnReadyTimeoutMs())
|| !Objects.equals(old.spawnReadyPollMs(), fresh.spawnReadyPollMs())) {
changed.add("spawnReady*");
@@ -265,14 +462,101 @@ public final class ConfigRef implements Supplier<FleetConfig> {
}
/**
* Whether two versions of a profile would launch a peer identically. Compares every component
* the launcher reads at spawn; {@code weight}, {@code maxLoad} and {@code credentialId} are
* excluded because those are read live (by the placement policy and, for credentialId, by
* {@code CompositePeerLauncher}/the CB-578 stage B exhaustion sink) and really do take effect on
* the next spawn.
* Split keys whose value differs between the running config and the candidate — see the class
* doc's Split bullet (fleetd #330, extended for {@code fleet:} by fleetd #333). Unlike
* {@link #changedDeferredKeys}, this does not try to tell which sub-field moved for {@code
* health:} or {@code coordinator:}: any change to either gets the same fixed message, because
* the message already names both halves every time, so there is no "which half changed"
* question left for the caller to answer. {@code fleet:} is different on purpose — see below.
*
* <p>Package-private (not {@code private}) so {@code ConfigRefTopLevelReportingCoverageTest}
* can call it directly with a reflection-built {@code FleetConfig} pair, the same reason
* {@link #sameLaunchSettings} is package-private — see that test's class doc (fleetd #333). That
* test exists because membership in {@link #SPLIT_KEYS} proves nothing about this method on its
* own: fleetd #333 measured that dropping the {@code coordinator} branch out of this method
* while leaving {@code "coordinator"} in {@code SPLIT_KEYS} left the whole suite green except a
* hand-written {@code ConfigRefTest} case — neither {@code ConfigRefTopLevelCoverageTest} (it
* only reads the set) nor the "kept in step" assert below (it only checks the reported keys are
* a SUBSET of {@code SPLIT_KEYS}, never that every {@code SPLIT_KEYS} member has a branch here)
* would have caught it.
*/
private static boolean sameLaunchSettings(FleetConfig.Profile a, FleetConfig.Profile b) {
return Objects.equals(a.baseUrl(), b.baseUrl())
static List<String> changedSplitKeys(FleetConfig old, FleetConfig fresh) {
List<String> changed = new ArrayList<>();
if (!Objects.equals(old.health(), fresh.health())) {
changed.add("health: the monitor itself (enabled, interval, workingSuspectAfter) is "
+ "frozen at startup and needs a restart; the coverage status fleet_profiles "
+ "reports is read live and already applied");
}
if (!Objects.equals(old.coordinator(), fresh.coordinator())) {
changed.add("coordinator: the LeadMailbox connection (uri, uriEnv, selfId, prefetch) is "
+ "opened once and needs a restart; the broker URI env-var name kept out of a "
+ "member's environment is read live on every spawn and already applied");
}
// fleetd #333: unlike health/coordinator above, most of `fleet:` (architects, developers,
// reviewers, charters, tabLabel) is genuinely hot — ConfigRefTest.aHotChangeIsAppliedAndRead-
// ThroughGet and aCharterChangeIsHotAndReachesTheLiveConfig prove it reaches the live config
// with no restart note. Only fleet.leaders is frozen (Fleetd.java:281 reads
// cfg.fleet().leaders() off the startup snapshot to build both the LeadTabScanner's
// tab-label-to-name map, wired into CallerResolver.withLeadsAndMembers at Fleetd.java:620/624,
// and — when herdr answered — LeadLauncher(...).ensureLeads() at Fleetd.java:315, which
// auto-launches each lead up to its `instances` count; neither is rebuilt on reload). So this
// compares fleet.leaders alone, not the whole Fleet record: comparing the whole record would
// report "split" for a tabLabel-only or charters-only change that is actually fully hot,
// which is the over-claim mirror of the under-claim bug this class exists to prevent.
if (!Objects.equals(leadersOf(old), leadersOf(fresh))) {
changed.add("fleet: fleet.leaders (each lead's tab, workspace, cwd, profile and "
+ "instances count) is read once at startup to build the LeadTabScanner's "
+ "identity map and to auto-launch leads, and neither is rebuilt on reload, so a "
+ "lead added, removed, or given a new tab: label needs a restart — until then it "
+ "stays unrecognised, and a caller from its new tab resolves as a worker, not a "
+ "lead; the rest of fleet: (architects, developers, reviewers, charters, "
+ "tabLabel) is read live through the supplier on CompositePeerLauncher and "
+ "already applied");
}
// Kept in step with SPLIT_KEYS the same way changedColdKeys is kept in step with COLD_KEYS —
// every message here must be traceable to one of the split keys the class doc documents.
// NOTE what this does NOT prove, per the javadoc above: it does not catch a SPLIT_KEYS
// member with no branch above at all, only a branch whose message is mis-worded relative to
// the set. ConfigRefTopLevelReportingCoverageTest is what proves the former.
assert changed.stream().allMatch(m -> SPLIT_KEYS.stream().anyMatch(k -> m.startsWith(k + ":")))
: "a split entry was reported that does not start with a SPLIT_KEYS name: " + changed;
return changed;
}
/** {@code cfg.fleet().leaders()}, defensively, in case a caller hands in a non-defaulted config. */
private static Map<String, FleetConfig.Leader> leadersOf(FleetConfig cfg) {
return cfg.fleet() == null ? Map.of() : cfg.fleet().leaders();
}
/**
* {@link FleetConfig.Profile} record components deliberately left out of
* {@link #sameLaunchSettings} because they are read <em>live</em>, not baked in at spawn — see
* the class doc's <em>Hot</em> bullet. {@code weight} and {@code maxLoad} are read live by the
* placement policy on every spawn; {@code credentialId} is read live by
* {@code CompositePeerLauncher} and the CB-578 stage B exhaustion sink. Nothing else is
* excluded — see {@code sameLaunchSettingsComparesEveryProfileComponentOrExcludesIt} in
* {@code ConfigRefProfileCoverageTest}, which enumerates every {@code Profile} record component
* by reflection and fails the build if one is neither compared below nor named here.
*/
static final Set<String> LAUNCH_SETTINGS_EXCLUDED = Set.of("weight", "maxLoad", "credentialId");
/**
* Whether two versions of a profile would launch a peer identically.
*
* <p>This must compare every {@link FleetConfig.Profile} record component except the three in
* {@link #LAUNCH_SETTINGS_EXCLUDED}. That is not a claim this javadoc can make good on by
* itself — a javadoc saying "compares every component" is exactly what fleetd #323 found to be
* false for three fields (and a sibling method's field list, for a fourth). The actual
* guarantee comes from {@code ConfigRefProfileCoverageTest}: it enumerates every record
* component of {@code FleetConfig.Profile} by reflection, mutates each one not in
* {@code LAUNCH_SETTINGS_EXCLUDED} on a base profile, and asserts this method reports a
* difference — so a new component that is neither compared here nor added to
* {@code LAUNCH_SETTINGS_EXCLUDED} (with a reason) fails that test by name, rather than
* silently reporting "config reloaded" for a value the daemon never picked up.
*/
static boolean sameLaunchSettings(FleetConfig.Profile a, FleetConfig.Profile b) {
return Objects.equals(a.profile(), b.profile())
&& Objects.equals(a.baseUrl(), b.baseUrl())
&& Objects.equals(a.model(), b.model())
&& Objects.equals(a.configDir(), b.configDir())
&& Objects.equals(a.tokenEnv(), b.tokenEnv())
@@ -298,6 +582,16 @@ public final class ConfigRef implements Supplier<FleetConfig> {
// fleetd #201 Unit 5: errorPattern is compiled once into Fleetd.main's backend-error
// pattern map at startup (see BackendErrorPatternLookup wiring), the same way
// exhaustedPattern is — a reload never re-reads it either.
&& Objects.equals(a.errorPattern(), b.errorPattern());
&& Objects.equals(a.errorPattern(), b.errorPattern())
// fleetd #323 instance 1: ideProjectDir and ideOpenCommand are read at spawn off the
// same frozen profile map as ideMcpUrl above (ClaudeCodeLauncher.java:267/269,
// OpenCodeLauncher.java:474/480/486) and were missing from this comparison.
&& Objects.equals(a.ideProjectDir(), b.ideProjectDir())
&& Objects.equals(a.ideOpenCommand(), b.ideOpenCommand())
// fleetd #323 instance 1: autoCompactWindow is read at spawn the same way
// (ClaudeCodeLauncher.java:926, OpenCodeLauncher.java:650). Comparing it here only
// makes the reload REPORT that a restart is needed — it deliberately does not make
// autoCompactWindow take effect live, which is a separate, larger change.
&& Objects.equals(a.autoCompactWindow(), b.autoCompactWindow());
}
}
@@ -1,10 +1,12 @@
package dev.ltms.fleet.inject;
/**
* Notified when {@link CompletionResolver} actually delivers a typed backend-error classification
* to a waiting send (fleetd#201 / #227) — never on a race that lost. {@link CompletionResolver}
* calls this only after {@code Rendezvous.resolveFailure} returns {@code true} for that exact
* waiter, mirroring the win-only race rule {@link ExhaustionSink} already uses.
* Notified when {@link CompletionResolver} has a backend-error match at the start of a pane line,
* or both a match and its too-fast crash signature, for a waiting send (fleetd#201 / #227). A text
* match inside ordinary pane prose can be a member's report about an error, so it fails the send
* without notifying this sink.
* {@link CompletionResolver} calls this only after {@code Rendezvous.resolveFailure} returns
* {@code true} for that exact waiter, mirroring the win-only race rule {@link ExhaustionSink} uses.
*
* <p>The public send result is unchanged by this classification — it is still a failed send
* ({@code Rendezvous.Kind#FAILED}); this sink is the internal seam a later stage (fleetd#201 Unit
@@ -387,9 +387,10 @@ public final class CompletionResolver implements TurnListener {
// fleetd#164 (part 2) / fleetd#201: a scrape that read cleanly and produced content still
// isn't a real reply when that content is the backend's own rejection (e.g. an HTTP 400
// before the worker did any work). Classify it as a failure naming the member, rather than
// handing the caller a scrape that reads like a completed answer, and — only on the
// resolution that actually wins the race, mirroring the exhaustion sink above — notify the
// typed backend-error sink so a later stage can act on repeated failures.
// handing the caller a scrape that reads like a completed answer. A text match alone is not
// enough to notify the typed backend-error sink: this assistant block can be a member's
// normal prose about an error. A line that starts with the error match is stronger evidence;
// the too-fast path below also has its crash signature before it records a credential failure.
String backendError = firstMatchingLine(assistantBlock, backendErrorPatternOrFallback(target));
if (backendError != null) {
// Carry the whole scrape, not just the matched line. The pattern is a heuristic: a member
@@ -401,9 +402,9 @@ public final class CompletionResolver implements TurnListener {
if (rendezvous.resolveFailure(waiter, reason)) {
inFlight.remove(target, turn);
log.warn("failing send to {} via turn-stall fallback: {}", target, reason);
// fleetd#201 Unit 1: only on the resolution that actually won the race — a late
// duplicate must never double-count one backend failure.
backendErrorSink.onBackendError(target, backendError, reason);
if (startsWithBackendError(backendError, backendErrorPatternOrFallback(target))) {
backendErrorSink.onBackendError(target, backendError, reason);
}
}
return;
}
@@ -492,8 +493,9 @@ public final class CompletionResolver implements TurnListener {
if (rendezvous.resolveFailure(waiter, reason)) {
inFlight.remove(target, turn);
log.warn("failing send to {} via turn-stall fallback from the raw scrape: {}", target, reason);
// fleetd#201 Unit 1: only on the resolution that actually won the race.
backendErrorSink.onBackendError(target, backendError, reason);
if (startsWithBackendError(backendError, backendErrorPatternOrFallback(target))) {
backendErrorSink.onBackendError(target, backendError, reason);
}
}
return true;
}
@@ -540,10 +542,10 @@ public final class CompletionResolver implements TurnListener {
* {@link #MIN_TURN_NANOS} — a crash signature (e.g. a backend HTTP 400 before the worker did
* anything) that a bare {@code BUSY -> DONE} transition cannot be told apart from a genuinely
* fast completion. Runs the same backend-error classification the normal and raw-scrape paths
* apply, against whatever is on screen right now: a match is a typed failure that 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".
* 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".
*/
private void failTooFast(String target, InFlight turn, CompletableFuture<Rendezvous.Resolution> waiter,
long elapsedNanos) {
@@ -611,6 +613,28 @@ public final class CompletionResolver implements TurnListener {
return null;
}
/**
* True when the error pattern begins the matched pane line, rather than appearing in prose.
*
* <p>Leading terminal chrome is skipped first — box-drawing characters, bullets, gutter bars and
* spaces. #339 introduced this check with a bare {@code lookingAt}, and that rejected a genuine
* error line rendered as {@code "| 503 Service Unavailable: ..."}: the send still failed, but the
* credential outage was never recorded. That is the false negative #339's own invariant 3 called
* worse than the false positive it set out to fix — measured with a throwaway probe on the
* raw-scrape path, which is exactly the path whose comment says to expect leading chrome.
*
* <p>Skipping only a leading run of non-letter, non-digit characters keeps the fix's intent. A
* member's prose ({@code "I checked the retry path. An API Error: makes it back off."}) still
* does not match, because there the pattern sits after words, not after chrome.
*/
private static boolean startsWithBackendError(String line, Pattern pattern) {
int i = 0;
while (i < line.length() && !Character.isLetterOrDigit(line.charAt(i))) {
i++;
}
return pattern.matcher(line.substring(i)).lookingAt();
}
/**
* Coverage summary for the CB-578 stage A exhausted-pattern classification, logged at startup
* the way {@link dev.ltms.fleet.health.FleetHealthMonitor#coverage} is — so an operator can
@@ -145,8 +145,58 @@ public final class Injector {
return router != null ? router.agentsFor(target) : agents;
}
/** The result of trying to remove an undelivered message from the injector. */
public enum Cancellation {
CANCELLED,
DELIVERED,
NOT_DELIVERED
}
/**
* An identity handle for one queued delivery. It is the only value accepted by
* {@link #cancel(Delivery)}, so a caller cannot cancel a different message with the same target
* or text.
*/
public static final class Delivery {
private final Pending pending;
private Delivery(Pending pending) {
this.pending = pending;
}
public CompletableFuture<Void> completion() {
return pending.delivered;
}
}
/** A pending message and the future that completes when it has been delivered. */
private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
private static final class Pending {
enum State { QUEUED, DELIVERED, NOT_DELIVERED, CANCELLED }
final String target;
final String text;
final TurnToken token;
final CompletableFuture<Void> delivered;
volatile State state = State.QUEUED; // written under the owning Target monitor
Pending(String target, String text, TurnToken token, CompletableFuture<Void> delivered) {
this.target = target;
this.text = text;
this.token = token;
this.delivered = delivered;
}
String text() {
return text;
}
TurnToken token() {
return token;
}
CompletableFuture<Void> delivered() {
return delivered;
}
}
/** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -177,15 +227,47 @@ public final class Injector {
* <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
* target" and "queue the message" and orphan it in a target it just removed.
*/
public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
public Delivery enqueue(String target, String text, TurnToken token) {
CompletableFuture<Void> delivered = new CompletableFuture<>();
Pending p = new Pending(text, token, delivered);
Pending p = new Pending(target, text, token, delivered);
targets.compute(target, (_, existing) -> {
Target t = (existing != null) ? existing : new Target();
t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
return t;
});
return delivered;
return new Delivery(p);
}
/**
* Cancel this exact queued delivery. The target monitor serializes this operation with
* {@link #onStatus}: if delivery wins that race, this returns {@link Cancellation#DELIVERED}
* rather than claiming the message remained queued.
*/
public Cancellation cancel(Delivery delivery) {
Pending p = delivery.pending;
Target t = targets.get(p.target);
if (t == null) {
return cancellationOf(p);
}
synchronized (t) {
if (p.state != Pending.State.QUEUED || !t.queue.remove(p)) {
return cancellationOf(p);
}
p.state = Pending.State.CANCELLED;
if (isQuiescent(t)) {
targets.remove(p.target, t);
}
return Cancellation.CANCELLED;
}
}
private static Cancellation cancellationOf(Pending p) {
return p.state == Pending.State.DELIVERED ? Cancellation.DELIVERED : Cancellation.NOT_DELIVERED;
}
private static boolean isQuiescent(Target t) {
return t.queue.isEmpty() && !t.awaitingPickup && !t.awaitingCompletion
&& !t.postTurnPending && !t.awaitingPostTurnPickup && !t.postTurnObserved;
}
/**
@@ -274,6 +356,7 @@ public final class Injector {
try {
agentsFor(target).send(target, p.text());
t.queue.poll();
p.state = Pending.State.DELIVERED;
t.awaitingPickup = true;
t.awaitingCompletion = true;
t.turnObserved = false;
@@ -283,6 +366,7 @@ public final class Injector {
// Delivery failed at herdr; drop the poisoned message and surface it
// rather than blocking the queue behind it.
t.queue.poll();
p.state = Pending.State.NOT_DELIVERED;
sent = p;
sendError = e;
}
@@ -293,6 +377,9 @@ public final class Injector {
// fail every queued message and release the target (CB-114) instead of
// polling it indefinitely with the caller's future never completing.
notReady = new ArrayList<>(t.queue);
for (Pending pending : notReady) {
pending.state = Pending.State.NOT_DELIVERED;
}
log.warn("readiness grace for {} expired after {} polls ({}s): target never "
+ "became deliverable, so failing {} queued message(s) that never "
+ "reached its pane",
@@ -340,8 +427,7 @@ public final class Injector {
// Reclaim the entry once the worker is fully quiescent (nothing queued, no pickup or
// completion awaited), so the map cannot grow without bound across short-lived workers.
if (t.queue.isEmpty() && !t.awaitingPickup && !t.awaitingCompletion
&& !t.postTurnPending && !t.awaitingPostTurnPickup && !t.postTurnObserved) {
if (isQuiescent(t)) {
targets.remove(target, t);
}
}
@@ -432,6 +518,9 @@ public final class Injector {
boolean hadDeliveredTurn;
synchronized (t) {
pending = new ArrayList<>(t.queue);
for (Pending p : pending) {
p.state = Pending.State.NOT_DELIVERED;
}
t.queue.clear();
hadDeliveredTurn = t.awaitingCompletion;
t.awaitingCompletion = false;
@@ -90,13 +90,27 @@ public final class ConnectionIdentity {
* {@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.
* <p><strong>What excluding an address costs, stated as it is today.</strong> This paragraph
* used to say that narrowing this range turned a worker into the lead, and that widening the
* check was what closed the hole. That was true only while there were <em>two</em> definitions
* that disagreed: {@code ConnectionIdentity} skipped the identity lookup for {@code 127.0.0.2}
* while {@code CallerResolver} read the same address as loopback and granted the primary role.
* #305 removed the second copy, and with one shared definition the old sentence no longer holds.
*
* <p>Measured on 2026-09-04 by narrowing this method back to exactly {@code 127.0.0.1} and
* running {@code CallerResolverTest} and {@code ConnectionIdentityTest}: a caller from
* {@code 127.0.0.2} then resolves to {@code ANONYMOUS}, not {@code PRIMARY} — for a worker
* ({@code aWorkerOnAnyLoopbackSourceAddressIsStillAWorkerNotThePrimary}) and for a non-worker
* ({@code aNonWorkerOnAnyLoopbackSourceAddressIsStillThePrimary}) alike. Excluding an address
* now <em>refuses</em> its caller; it does not promote one.
*
* <p>So keep the whole range, but for the plain reason: a genuine worker or primary that
* connects from {@code 127.0.0.2} must be identifiable at all, and narrowing this predicate
* locks it out. That is an outage, and an outage is the direction to fail in — which is exactly
* why the range must not be narrowed casually and also why doing so is no longer a security
* hole. This predicate still does not decide whether a caller is trusted; it decides whether the
* caller's identity is <em>resolved at all</em>. What makes an unresolved caller safe is
* {@link Caller#resolved()} (#317), not this method.
*/
public static boolean isLoopback(String addr) {
if (addr == null) {
@@ -385,9 +385,12 @@ public final class CompositePeerLauncher implements PeerLauncher {
}
}
// unreachable.size() counts DISTINCT profiles, not attempts (a HashSet dedupes a profile
// added twice) — say "distinct" so the count matches the sentence and the profile list that
// follows, rather than reading as a count of attempts made (fleetd #315).
throw new PeerUnreachableException(
"no reachable worker profile available after trying " + unreachable.size()
+ " candidate(s): " + String.join(", ", unreachable));
+ " distinct candidate(s): " + String.join(", ", unreachable));
}
/**
@@ -1665,30 +1665,59 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
/**
* Guards the {@code effectiveAllowed == null} branch of {@link #logCredentialGap} — the
* genuinely-unprotected report (deny-by-default, and the allow-list non-zsh fallback) — to one
* WARN per launcher instance, not one per spawn.
* genuinely-unprotected report (deny-by-default, and the allow-list non-zsh fallback) — AND
* the {@code effectiveAllowed != null} / {@code keptByDerivedList} branch, the allow-list case
* where a name is in the gap but the derived allow-list keeps it anyway. Both branches log the
* same severity (WARN) about the same fact — a name genuinely reaching a member pane
* unprotected — so they share this one guard, keyed per NAME rather than per launcher instance:
* each credential-shaped name that is ever reported unprotected gets exactly one WARN, however
* many spawns see it and whichever of the two branches first reports it.
*
* <p>fleetd #341: {@code memberCredentials} is a live, re-read-per-spawn supplier, so the
* policy — and so the gap's actual member names — can change between two spawns on the same
* launcher. Before this fix the guard was a single {@code AtomicBoolean} tripped by either
* branch: spawn 1 could warn about name A and trip the flag, and a later spawn's gap containing
* a different name B would never be reported, even though B is just as unprotected as A was.
* {@code AtomicBoolean} could not express "once per distinct name" at all — only "once, ever,
* for whichever name got there first" — so this is a {@code Set<String>} guard instead, the
* same shape {@link OpenCodeLauncher#modelCheckSkippedWarned} already uses for its own
* once-per-distinct-thing WARN. {@link #add}'s return value (true only the first time a name is
* added) is what turns "log the whole gap" into "log only the names never warned about before".
*
* <p>Bounded by construction: every name added here first passed {@link
* #CREDENTIAL_SHAPED_NAME}'s filter over {@link #hostEnvNames}, i.e. it is an actual
* environment variable name from the daemon's own process — a small, OS-bounded set (the host
* environment has, in practice, tens to a few hundred entries), not an attacker- or
* request-controlled input. So this set cannot grow past "however many distinct credential-
* shaped names this host's environment has ever held across this launcher's lifetime," which is
* effectively fixed for the life of one daemon process — no separate cap is needed.
*
* <p>CB-633 follow-up (#192): kept SEPARATE from {@link #allowListGapLogged} on purpose.
* {@code memberCredentials} is a live, re-read-per-spawn supplier, so the policy can change
* between two spawns on the same launcher. A single shared flag would let a harmless allow-list
* INFO on spawn 1 permanently suppress the real deny-by-default WARN a later spawn deserves —
* the report that matters most getting hidden by the report that doesn't. Two flags mean each
* report kind fires exactly once, independent of what the other kind already logged.
* report kind (WARN vs. INFO) fires independently of what the other kind already logged; within
* the WARN kind itself, the set above further separates by name, for the same reason.
*/
private final AtomicBoolean unprotectedGapLogged = new AtomicBoolean();
private final Set<String> unprotectedGapNamesWarned = ConcurrentHashMap.newKeySet();
/**
* Guards the {@code effectiveAllowed != null} branch of {@link #logCredentialGap} — the
* allow-list-scrub-covered report — to one INFO per launcher instance. See {@link
* #unprotectedGapLogged}'s javadoc for why this is a separate flag rather than a shared one.
* #unprotectedGapNamesWarned}'s javadoc for why this is a separate flag rather than a shared
* one; unlike that guard it stays a per-instance {@code AtomicBoolean}, not a per-name set —
* fleetd #341 fixed the WARN-vs-WARN suppression, not this INFO's own one-shot shape, which was
* not reported as broken and is out of that ticket's scope.
*/
private final AtomicBoolean allowListGapLogged = new AtomicBoolean();
/**
* fleetd #185 stage 2: guards {@link #warnUnknownMemberEnvironment} to one WARN per launcher
* instance, not one per spawn — the same one-per-instance shape as {@link #unprotectedGapLogged}
* and {@link #allowListGapLogged}, kept as its own flag for the same reason those two are split:
* this mode is orthogonal to which of the other two branches would otherwise have fired.
* instance, not one per spawn — the same one-shot shape {@link #unprotectedGapNamesWarned} and
* {@link #allowListGapLogged} guard their own branches with, kept as its own flag for the same
* reason those two are split: this mode is orthogonal to which of the other two branches would
* otherwise have fired.
*/
private final AtomicBoolean unknownMemberEnvironmentWarned = new AtomicBoolean();
@@ -1816,14 +1845,21 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
List<String> blankedByScrub = gap.stream()
.filter(name -> !MemberEnvAllowList.keeps(effectiveAllowed, name))
.toList();
if (!keptByDerivedList.isEmpty() && unprotectedGapLogged.compareAndSet(false, true)) {
// fleetd #341: filter to names this guard has never warned about before — not just
// "isEmpty" on the whole branch — so a name this spawn's gap shares with an EARLIER
// spawn's (already-warned) gap does not re-print, while a name unique to THIS gap still
// does, whichever of the two WARN branches reported it first.
List<String> newlyUnprotected = keptByDerivedList.stream()
.filter(unprotectedGapNamesWarned::add)
.toList();
if (!newlyUnprotected.isEmpty()) {
log.warn("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
+ "known: nor allow: — the derived allow-list keeps them anyway (a profile's "
+ "gitTokenEnv/gitHostEnv/tokenEnv/env: names one, or this spawn injects it), "
+ "so every member pane inherits them UNBLOCKED — {}. Add each to "
+ "memberCredentials.known (or .allow if a member legitimately needs it), or "
+ "remove it from whatever profile setting derives it in.",
keptByDerivedList.size(), keptByDerivedList);
newlyUnprotected.size(), newlyUnprotected);
}
if (!blankedByScrub.isEmpty() && allowListGapLogged.compareAndSet(false, true)) {
log.info("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
@@ -1836,12 +1872,18 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
/** The deny-by-default (and allow-list non-zsh fallback) WARN — unchanged byte-for-byte by #192. */
private void warnGapUnprotected(List<String> gap) {
if (unprotectedGapLogged.compareAndSet(false, true)) {
// fleetd #341: same "only the names never warned before" filter as the sibling branch in
// logCredentialGap above — see unprotectedGapNamesWarned's javadoc. Both branches share
// this one guard because both report the exact same fact (a name reaching a member pane
// unprotected) at the exact same severity; keying it by name is what lets a later spawn's
// DIFFERENT name still get its own WARN after an earlier spawn's already fired.
List<String> newlyUnprotected = gap.stream().filter(unprotectedGapNamesWarned::add).toList();
if (!newlyUnprotected.isEmpty()) {
log.warn("memberCredentials gap: {} credential-shaped env var name(s) are on neither "
+ "known: nor allow: — every member pane inherits them UNBLOCKED — {}. "
+ "Add each to memberCredentials.known (blocked by default) or .allow "
+ "(if a member legitimately needs it).",
gap.size(), gap);
newlyUnprotected.size(), newlyUnprotected);
}
}
@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
/**
* AMQP-backed {@link ReplyInbox} (CB-307 Stage 2): genuine cross-restart durability behind the same
@@ -93,8 +94,23 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
private final Channel channel;
/** All channel operations (publish/declare/ack/cancel) serialize on this — a Channel is not thread-safe. */
private final Object channelLock = new Object();
/** target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself. */
/**
* target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself.
*
* <p><strong>CB-318 tombstone.</strong> The value {@link #RELEASED} is a reserved sentinel: it
* marks a target whose {@link #release} has already run, so {@link #deliverCallback} can tell a
* delivery landing after release() apart from a fresh target it has never seen. See both methods'
* javadoc for why a plain {@code held.remove(target)} is not enough.
*/
private final ConcurrentHashMap<String, LinkedHashMap<String, Held>> held = new ConcurrentHashMap<>();
/**
* CB-318 sentinel stored in {@link #held} for a target whose {@link #release} has already run.
* Never mutated — every read site compares it by reference ({@code ==}) before touching it as a
* map, because it is a single object shared across every released target and calling a mutator on
* it would corrupt state for all of them.
*/
private static final LinkedHashMap<String, Held> RELEASED = new LinkedHashMap<>();
/** Targets whose queue is declared and consumer is running, mapped to their broker consumer tag. */
private final ConcurrentHashMap<String, String> consumerTags = new ConcurrentHashMap<>();
@@ -201,6 +217,12 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
channel.queueDeclare(queue, true, false, false, null); // durable, non-exclusive, keep on idle
String tag = channel.basicConsume(queue, false, deliverCallback(target), _ -> { });
consumerTags.put(target, tag);
// CB-318: drop a stale RELEASED tombstone from a prior ownership of this same target
// string, so a delivery under this fresh consumer is held normally instead of being
// nacked forever by deliverCallback's RELEASED check. Safe to do here, still under
// channelLock: no delivery for the consumer tag just registered above can reach
// deliverCallback before this basicConsume call returns.
held.remove(target, RELEASED);
log.debug("AMQP inbox owns queue {} for target {}", queue, target);
} catch (IOException e) {
throw new IllegalStateException("cannot own queue " + queue, e);
@@ -243,6 +265,32 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
* later connection drop even though this release did not manage to requeue it immediately. A
* failed {@code basicCancel} still throws, unchanged from before this fix — that failure means
* the consumer may still be attached, so best-effort requeue is not attempted underneath it.
*
* <p><strong>CB-318: {@code held.remove(target)} alone leaves a second window open.</strong> The
* bullet above already explains why cancelling first does not save a tag from going stale — but
* that only accounts for a delivery landing before this method starts touching {@link #held}.
* {@code basicCancel} stops <em>new</em> dispatches; it does not flush one already handed to the
* consumer work pool. So a delivery can still land on that pool's thread and reach
* {@link #deliverCallback} at any point during, or after, this method's body — and a plain
* {@code held.remove(target)} does nothing to stop it: {@code deliverCallback}'s
* {@code computeIfAbsent} finds the key gone and happily creates a brand-new map under it, which
* this method — already past its {@code remove} — never looks at again. That entry then sits
* delivered-but-unacked on {@link #channel} until the whole inbox closes: never requeued, never
* redelivered, and {@link #peek} is never called again for a target nothing owns any more.
*
* <p>The fix is {@link #held}{@code .compute(target, ...)} instead of {@code remove}: it takes
* whatever was held (to nack, same as before) and, in the same atomic step, leaves the
* {@link #RELEASED} tombstone behind instead of an absent key. {@code computeIfAbsent} and
* {@code compute} calls for the same key are mutually exclusive in {@link ConcurrentHashMap} —
* whichever of this call and a concurrent {@code deliverCallback} runs first is fully visible to
* the other, with no gap between them. So a delivery that loses the race sees a real map here and
* gets nacked by the loop below, same as always; a delivery that wins the race (runs first) is
* itself nacked by that same loop, once it settles into {@code held}. A delivery that arrives once
* this method has stored {@link #RELEASED} finds it via {@code computeIfAbsent} and refuses itself
* — see {@link #deliverCallback}. Either way nothing is silently retained forever, satisfying the
* ticket's invariant against dropping a message. This closes the window rather than merely
* narrowing it — correctness does not depend on how much time elapses between the swap and this
* method returning.
*/
@Override
public void release(String target) {
@@ -255,8 +303,13 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
throw new IllegalStateException("cannot cancel consumer for " + target, e);
}
}
var perTarget = held.remove(target);
if (perTarget != null) {
AtomicReference<LinkedHashMap<String, Held>> previouslyHeld = new AtomicReference<>();
held.compute(target, (_, v) -> {
previouslyHeld.set(v);
return RELEASED;
});
var perTarget = previouslyHeld.get();
if (perTarget != null && perTarget != RELEASED) {
synchronized (perTarget) {
for (Held h : perTarget.values()) {
try {
@@ -326,7 +379,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
@Override
public List<InboxMessage> peek(String target) {
var perTarget = held.get(target);
if (perTarget == null) {
if (perTarget == null || perTarget == RELEASED) {
return List.of();
}
synchronized (perTarget) {
@@ -337,7 +390,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
@Override
public void ack(String target, String msgId) {
var perTarget = held.get(target);
if (perTarget == null) {
if (perTarget == null || perTarget == RELEASED) {
return;
}
Held h;
@@ -370,6 +423,20 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
}
String content = new String(delivery.getBody(), StandardCharsets.UTF_8);
var perTarget = held.computeIfAbsent(target, _ -> new LinkedHashMap<>());
if (perTarget == RELEASED) {
// CB-318: release() already ran for this target and left the RELEASED tombstone in
// held (see release()'s javadoc) — computeIfAbsent() is guaranteed to see it rather
// than recreate a fresh map, because ConcurrentHashMap serializes compute/
// computeIfAbsent calls for the same key against each other. Refuse the delivery
// instead of holding it somewhere release() will never look at again: requeue it, the
// same way release() nacks its own held entries, so a later owner (or a connection
// drop) can still recover it. This does not need channelLock across a broker round
// trip — basicNack, like the duplicate-ack case just below, does not wait for one.
synchronized (channelLock) {
channel.basicNack(tag, false, true);
}
return;
}
boolean duplicate;
synchronized (perTarget) {
if (perTarget.containsKey(msgId)) {
@@ -466,8 +466,18 @@ public final class MessageService {
if (candidates.size() == 1) {
Task orphan = candidates.get(0);
if (orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
if (orphan.turnId != null) {
asyncTasksByTurn.remove(orphan.turnId, orphan);
// fleetd #329 (F3): read orphan.turnId once. It used to be read twice, under no
// lock — once for this null check, once as the removal key — the same double-read
// shape fleetd #324 fixed in finishAsyncTask. clearAsyncQuestion's unlocked
// forgetTurn=true path (ask()'s timeout cleanup) can null the field between the two
// reads; capturing it once removes the torn read here too.
String turnId = orphan.turnId;
if (turnId != null) {
if (replyOrphanTurnIdRaceHookForTest != null) {
// Test-only (fleetd #329, F3): see the field's own javadoc.
replyOrphanTurnIdRaceHookForTest.run();
}
asyncTasksByTurn.remove(turnId, orphan);
}
count(FleetMetrics.REPLIES, "path", "async-recovered");
return true; // the ticket itself took it — no inbox stranding at all
@@ -853,16 +863,26 @@ public final class MessageService {
if (onAccepted != null) {
onAccepted.run();
}
CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
Injector.Delivery delivery = injector.enqueue(target, content, token);
try {
Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
} catch (TimeoutException e) {
boolean wasDelivered = delivered.isDone() && !delivered.isCompletedExceptionally();
boolean wasDelivered = delivery.completion().isDone()
&& !delivery.completion().isCompletedExceptionally();
if (!wasDelivered) {
if (timeoutCancellationRaceHookForTest != null) {
// Test-only (fleetd #345): see the field's own javadoc.
timeoutCancellationRaceHookForTest.run();
}
// The target monitor makes cancellation atomic with onStatus picking this
// Pending up. If pickup won, report TIMED_OUT_WORKING because the text landed.
wasDelivered = injector.cancel(delivery) == Injector.Cancellation.DELIVERED;
}
log.debug("send to {} timed out (delivered={})", target, wasDelivered);
if (!wasDelivered) {
// CB-640: still sitting in the injector's queue, waiting for the member to
// go idle — record the fact for fleet health (see queuedDeliveries).
// CB-640: record that delivery did not happen for fleet health (see
// queuedDeliveries). The exact Pending was cancelled, so it cannot arrive later.
queuedDeliveries.put(target, Boolean.TRUE);
}
return recorded(new Reply(
@@ -1002,13 +1022,49 @@ public final class MessageService {
// Measured when #282 was merged: this guard is DEFENCE IN DEPTH, not the thing
// that makes the chained ask work. ask() calls markAsyncQuestion (:860) before
// resolveQuestion (:861), so by the time this thread wakes, the task has already
// moved to the new turnId and finishAsyncTask(oldTurnId, ...) finds nothing. Removing
// this guard alone leaves the test green. Keep it anyway: it mirrors sendAsync's
// sibling guard, and that sibling's own comment (:1017) warns the two orderings are
// not something to rely on. Do NOT delete it as dead code without re-checking that
// ordering, and do not treat it as the sole protection either.
if (result.outcome() != Outcome.QUESTION) {
finishAsyncTask(turnId, result);
// moved to the new turnId — this outcome check, not a Task lookup, is what tells the
// two cases apart (see fleetd #329 below). Removing this guard alone leaves the test
// green. Keep it anyway: it mirrors sendAsync's sibling guard, and that sibling's own
// comment (:1017) warns the two orderings are not something to rely on. Do NOT delete
// it as dead code without re-checking that ordering, and do not treat it as the sole
// protection either.
//
// fleetd #329 (F1): complete the SAME Task object this method already looked up at
// :991, rather than re-resolving it from turnId a second time. The old
// finishAsyncTask(turnId, result) did its own asyncTasksByTurn.get(turnId) here, and
// that second lookup races ask()'s own timeout path: ask()'s ticket.answer().get(...)
// can time out (or lose that exact race) at essentially the same instant this method's
// rendezvous.answerAsk(turnId, ...) above already succeeded, running
// markAskTimedOut + clearAsyncQuestion(turnId, true) with no lock at all — which
// forgets turnId (removes it from asyncTasksByTurn, nulls Task.turnId) before this
// thread ever gets here. The worker's real reply then arrived, this method's own wait
// woke up with it, and the by-turnId lookup found nothing: the async ticket's future
// was never completed, so fleet_poll{ticket} stayed PENDING forever even though
// answer() itself correctly returned REPLIED. Reusing the reference captured at :991
// — before that race window opens — sidesteps the second lookup entirely: it is the
// identical Task whatever asyncTasksByTurn or Task.turnId say by the time we reach
// this point, so finishAsyncTask(Task, Reply) can still complete its future and detach
// it using whatever turnId it now reads. The #282 chained-ask case is unaffected
// because it is still gated purely by result.outcome() == QUESTION above, which does
// not depend on this lookup — widening what "task" means here cannot complete a ticket
// the chained ask deliberately left open.
//
// A null task is NOT only "this was never an async ticket". That reading was in this
// comment when #329 merged and it is wrong. A genuine async ticket also lands here
// with task == null, because ask()'s timeout path runs clearAsyncQuestion(turnId,
// true) — which drops the asyncTasksByTurn entry — in its catch block, while
// rendezvous.closeAsk(turnId) runs later, in its finally. Between those two the ask
// is still answerable but the map entry is already gone, so the lookup at :991
// returns null and this ticket is never completed. Measured on 2026-09-04: a probe
// firing only that first half before answer() runs printed
// "answer=REPLIED phase=PENDING reply=null" — the same stranded ticket #329 set out
// to fix, one step earlier in the same race. The probe used forgetTurnForTest, which
// omits ask()'s markAskTimedOut; that cannot change the outcome, because askTimedOut
// is read only by askAnsweredAsyncTasks, and reply() never reaches it while this
// method's own waiter is live. So #329 narrows this window rather than closing it.
// Open as fleetd #334 — do not read this guard as complete.
if (result.outcome() != Outcome.QUESTION && task != null) {
finishAsyncTask(task, result);
}
return result;
} catch (TimeoutException e) {
@@ -1059,7 +1115,7 @@ public final class MessageService {
// this fires exactly once, from whichever path completes it: finishAsyncTask(task, result)
// below on any non-QUESTION outcome of send() — a worker's fleet_reply, the CB-106
// completion fallback, a CB-109 wedge, TIMED_OUT, BUSY, or BACKEND_EXHAUSTED — the same
// finishAsyncTask reached via answer()'s finishAsyncTask(turnId, result) once a QUESTION
// finishAsyncTask reached via answer()'s own finishAsyncTask(task, result) once a QUESTION
// is resolved, completeExceptionally(t) just below when send() itself throws, or a CB-516
// abandon() on teardown. Without this, MessageService.reply's rendezvous fast path (the
// one an async ticket always takes) never told the push loop anything happened — see the
@@ -1079,7 +1135,22 @@ public final class MessageService {
finishAsyncTask(task, result);
}
} catch (Throwable t) {
task.future.completeExceptionally(t);
// fleetd #329 (F2): finishAsyncTask above already completes task.future — on its very
// first line — before doing anything else, so anything that throws afterward (inside
// finishAsyncTask's own cleanup, or from a future addition to this try block) lands
// here with the future already resolved. completeExceptionally on an already-completed
// future is a silent no-op: it returns false and does nothing, so without the check
// below the exception simply vanished — no log, no metric, nothing. Measured (see the
// ticket): temporarily reintroducing the fleetd #324 NPE reproduced 19 real exceptions
// on the ordinary path, with 74/74 tests staying green and not one log line produced.
// Do not stop completing the future first — finishAsyncTask completing before it
// cleans up is what makes a late failure harmless to the ticket's own result — only
// add the missing visibility for the case where that step, or whatever ran after it,
// has already lost the race to report through the future.
if (!task.future.completeExceptionally(t)) {
log.error("async send {} -> {} threw after its ticket was already resolved",
ticket, target, t);
}
}
});
pruneTerminalTickets();
@@ -1230,20 +1301,124 @@ public final class MessageService {
}
}
/** Complete and detach an async ticket after its worker's actual terminal reply. */
/**
* Complete and detach an async ticket after its worker's actual terminal reply.
*
* <p><strong>fleetd #324.</strong> {@code task.turnId} is read into {@code turnId} exactly once.
* It used to be read twice — once for the null check, once as the removal key — and {@code
* volatile} makes each of those reads individually fresh but does not make the pair atomic.
* {@link #answer} calls this while holding {@code sessionLocks} for the target; {@link #ask}'s
* own timeout path calls {@link #clearAsyncQuestion} (which nulls {@link Task#turnId}) under no
* lock at all. When that unlocked null-out landed between the two reads here, the second read saw
* {@code null} and {@code asyncTasksByTurn.remove(null, task)} threw {@code NullPointerException}
* on the lead's own {@code answer()} call — even though {@code task.future.complete(result)} on
* the line above had already run, so the answer was in fact delivered. Capturing the field once
* removes the torn read; see the ticket for why the wider asymmetry between the locked and
* unlocked sides is not fixed by this alone.
*/
private void finishAsyncTask(Task task, Reply result) {
task.future.complete(result);
if (task.turnId != null) {
asyncTasksByTurn.remove(task.turnId, task);
if (afterFinishAsyncTaskCompleteHookForTest != null) {
// Test-only (fleetd #329, F2): see the field's own javadoc.
afterFinishAsyncTaskCompleteHookForTest.run();
}
String turnId = task.turnId;
if (turnId != null) {
if (finishAsyncTaskRaceHook != null) {
// Test-only (fleetd #324): see the field's own javadoc.
finishAsyncTaskRaceHook.run();
}
asyncTasksByTurn.remove(turnId, task);
}
}
/** Complete the async ticket correlated to a specific answered turn. */
private void finishAsyncTask(String turnId, Reply result) {
Task task = asyncTasksByTurn.get(turnId);
if (task != null) {
finishAsyncTask(task, result);
}
/**
* Null in production; test seam for fleetd #324 — invoked from {@link #finishAsyncTask(Task,
* Reply)} right after {@code task.turnId}'s null-check passes and before the (now-local) value is
* used for the removal. A test installs this to force, deterministically, the exact interleaving
* that a real race between this method and {@link #ask}'s unlocked timeout cleanup can otherwise
* only produce by chance: firing it here reproduces "the field went null between the check and the
* use" against the pre-fix code, and demonstrates the fix tolerates it (the captured local is used
* unconditionally, so a hook that nulls the field afterward cannot affect this call).
*/
private volatile Runnable finishAsyncTaskRaceHook;
/**
* Test-only (fleetd #324): install {@link #finishAsyncTaskRaceHook}. Package-private so the test,
* in the same package, can reach it without widening any production API.
*/
void setFinishAsyncTaskRaceHookForTest(Runnable hook) {
this.finishAsyncTaskRaceHook = hook;
}
/**
* Null in production; test seam for fleetd #329 (F2) — invoked from {@link #finishAsyncTask(Task,
* Reply)} unconditionally, immediately after {@code task.future.complete(result)} runs (before
* {@link Task#turnId} is even read, so it fires regardless of whether this task was ever asked).
* A test installs this to force an exception into exactly the shape fleetd #329 identified:
* something throws inside {@link #sendAsync}'s executor task after the async ticket's future is
* already resolved, so the surrounding {@code catch (Throwable t)} can only report failure through
* {@code completeExceptionally} — a silent no-op on an already-completed future. Engineering a
* real exception to land in that exact post-completion window is what the ticket itself had to do
* by temporarily deleting a production guard (fleetd #324's {@code turnId != null} check); this
* hook drives the identical shape deterministically instead.
*/
private volatile Runnable afterFinishAsyncTaskCompleteHookForTest;
/**
* Test-only (fleetd #329, F2): install {@link #afterFinishAsyncTaskCompleteHookForTest}.
* Package-private so the test, in the same package, can reach it without widening any production
* API.
*/
void setAfterFinishAsyncTaskCompleteHookForTest(Runnable hook) {
this.afterFinishAsyncTaskCompleteHookForTest = hook;
}
/**
* Null in production; test seam for fleetd #345 — invoked in {@link #send}'s timeout path after
* {@link Injector.Delivery#completion()} reports incomplete and before {@link Injector#cancel}
* takes the target monitor. A test installs this to make {@code onStatus} pick the exact queued
* delivery up in that window, so {@code cancel} returns {@link Injector.Cancellation#DELIVERED}.
* This deterministically covers the caller's need to use that result rather than relying on a
* timing-sensitive real race.
*/
private volatile Runnable timeoutCancellationRaceHookForTest;
/**
* Test-only (fleetd #345): install {@link #timeoutCancellationRaceHookForTest}. Package-private
* so the test, in the same package, can reach it without widening any production API.
*/
void setTimeoutCancellationRaceHookForTest(Runnable hook) {
this.timeoutCancellationRaceHookForTest = hook;
}
/**
* Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
* the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
* value is used as the {@code asyncTasksByTurn} removal key. Mirrors {@link
* #finishAsyncTaskRaceHook} exactly, for the structurally identical double-read fleetd #324 fixed
* in {@link #finishAsyncTask}: a test installs this to force, deterministically, {@code
* clearAsyncQuestion}'s unlocked {@code forgetTurn=true} path nulling {@link Task#turnId} in that
* exact window, and to confirm the single-read fix tolerates it (the captured local is used
* unconditionally, so a hook that nulls the field afterward cannot affect this call).
*/
private volatile Runnable replyOrphanTurnIdRaceHookForTest;
/**
* Test-only (fleetd #329, F3): install {@link #replyOrphanTurnIdRaceHookForTest}. Package-private
* so the test, in the same package, can reach it without widening any production API.
*/
void setReplyOrphanTurnIdRaceHookForTest(Runnable hook) {
this.replyOrphanTurnIdRaceHookForTest = hook;
}
/**
* Test-only (fleetd #324): run the exact production cleanup {@link #ask}'s own timeout path runs
* unlocked — {@link #clearAsyncQuestion(String, boolean)} with {@code forgetTurn=true} — so a test
* can reproduce that specific mutation instead of hand-rolling an approximation of it.
*/
void forgetTurnForTest(String turnId) {
clearAsyncQuestion(turnId, true);
}
/** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
@@ -5,10 +5,14 @@ import java.util.List;
/**
* Backward-compatible placement: an unqualified spawn always resolves to the configured default
* profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps and
* reachability so that a pre-existing config behaves identically after upgrade.
* profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps
* ({@code maxLoad}) so that a pre-existing config behaves identically after upgrade — capacity
* gating for automatic placement is deliberately out of scope for {@code fixed}, exactly as it
* always has been. Reachability is a narrower exception (fleetd #315, below): a profile is never
* checked for reachability up front, only skipped once it has already failed in <em>this same</em>
* spawn call's retry loop — see the unreachable case below.
*
* <p>Three exceptions walk past the default instead of returning it unconditionally:
* <p>Four exceptions walk past the default instead of returning it unconditionally:
* <ul>
* <li>Quarantine (CB-578 stage B): a quarantined default is a credential that just refused on
* a usage limit, not a transient capacity or reachability concern.
@@ -16,13 +20,21 @@ import java.util.List;
* ({@code BackendOutagePolicy}) — a separate, shorter-lived source from quarantine. When a
* profile is both quarantined and cooling off, only the quarantine reason is reported
* (exhaustion takes priority), matching {@code CompositePeerLauncher}'s explicit-spawn order.
* <li>Unreachable (fleetd #315): {@code CompositePeerLauncher.spawn} retries a failed candidate
* on the next one and rebuilds the {@link PlacementContext} so {@code ctx.unreachable()}
* names every profile that already failed with {@code PeerUnreachableException} in this same
* call. Without this check {@code select} kept handing back the same dead default forever —
* the retry loop's own comment says "so the policy excludes this profile", and this is what
* makes that true for {@code fixed} too, matching {@code weighted}/{@code round-robin}
* (both filter on {@code ctx.unreachable()} via {@link PlacementPolicyUtil#available}).
* <li>Weight 0 (CB-554): {@code fixed} is still automatic selection, so a profile the operator
* marked "never auto-select me" ({@code weight <= 0}) must be skipped here exactly as
* {@code weighted}/{@code round-robin} skip it — an explicit {@code fleet_spawn} naming
* the profile is unaffected, only this automatic fallback walk.
* </ul>
* A fleet where nothing is ever quarantined, cooling off, or weight-0 never exercises any of these
* paths, so today's behaviour is unchanged.
* A fleet where nothing is ever quarantined, cooling off, unreachable, or weight-0 never exercises
* any of these paths, so today's behaviour is unchanged — in particular, the very first selection
* of a spawn call always sees an empty {@code unreachable} set, so the first choice is untouched.
*/
final class FixedPlacementPolicy implements PlacementPolicy {
@@ -30,12 +42,12 @@ final class FixedPlacementPolicy implements PlacementPolicy {
public PlacementCandidate select(PlacementContext ctx) {
String d = ctx.defaultProfile();
if (d != null && !d.isBlank() && !ctx.quarantined().contains(d) && !ctx.coolingOff().contains(d)
&& !weightExcluded(ctx, d)) {
&& !ctx.unreachable().contains(d) && !weightExcluded(ctx, d)) {
return new PlacementCandidate(d, null, 1.0f, null);
}
for (PlacementCandidate c : ctx.candidates()) {
if (!ctx.quarantined().contains(c.profile()) && !ctx.coolingOff().contains(c.profile())
&& !c.excluded()) {
&& !ctx.unreachable().contains(c.profile()) && !c.excluded()) {
return new PlacementCandidate(c.profile(), null, c.weight(), c.maxLoad());
}
}
@@ -44,8 +56,9 @@ final class FixedPlacementPolicy implements PlacementPolicy {
// Exhaustion quarantine takes priority: reported only when quarantine is absent, so the
// message never claims "cooling off" for a profile that is really backend-exhausted.
boolean dCoolingOff = !dQuarantined && ctx.coolingOff().contains(d);
boolean dUnreachable = ctx.unreachable().contains(d);
boolean dWeightExcluded = weightExcluded(ctx, d);
if (dQuarantined || dCoolingOff || dWeightExcluded) {
if (dQuarantined || dCoolingOff || dUnreachable || dWeightExcluded) {
List<String> reasons = new ArrayList<>();
if (dQuarantined) {
reasons.add("is quarantined (backend exhausted)");
@@ -53,6 +66,9 @@ final class FixedPlacementPolicy implements PlacementPolicy {
if (dCoolingOff) {
reasons.add("is cooling off after repeated backend errors");
}
if (dUnreachable) {
reasons.add("is unreachable");
}
if (dWeightExcluded) {
reasons.add("has weight 0 (excluded from automatic selection)");
}
@@ -62,7 +78,7 @@ final class FixedPlacementPolicy implements PlacementPolicy {
}
if (!ctx.candidates().isEmpty()) {
throw new PlacementException("all worker profiles are excluded from automatic "
+ "selection (quarantined, cooling off, or weight-0)");
+ "selection (quarantined, cooling off, unreachable, or weight-0)");
}
throw new PlacementException("no worker profiles configured");
}
@@ -386,6 +386,31 @@ public final class SessionManager implements TurnListener {
// from the registry with no pane stop is an orphaned pane — a live terminal burning a fleet
// slot that no longer appears in the roster and can never be reclaimed.
launcher.stop(paneId);
if (removed != null && !preserveWorktree && removed.worktree() != null) {
// fleetd #316: the `dirty` read above ran while the worker could still write to this
// worktree, so a stale `false` must not be trusted to authorise the --force removal
// below. Re-read the worktree's state one more time, right here — immediately before
// the one step that would destroy it, and only on the path that is actually about to
// do that (invariant 4: no second unconditional `git status` on a release that already
// decided to preserve). By now `launcher.stop` has returned, so this read reflects
// whatever the worker managed to write up to and including its teardown, not whatever
// it had written at release-start time.
if (dirtyImmediatelyBeforeRemoval(removed)) {
preserveWorktree = true;
// The pre-stop snapshot above never ran for this session (the pre-stop read said
// clean), so this is the only chance to get the newly-discovered work into
// refs/wip/* rather than leaving the on-disk preserve as the sole copy. Best-effort,
// like every other snapshot attempt — trySnapshot logs and swallows its own failure.
String lateSnapshotRef = trySnapshot(removed, cause);
log.warn("release {} preserves worktree {} for pane={} terminal={}: it reported "
+ "clean before the pane stopped but dirty immediately before removal — the "
+ "worker wrote to it during teardown, and --force removing it now would "
+ "have destroyed that work{}",
cause, removed.worktree(), paneId, removed.terminalId(),
lateSnapshotRef == null ? "" : " (snapshotted to refs/wip/" + removed.branch()
+ " commit=" + lateSnapshotRef + ")");
}
}
if (removed != null && !preserveWorktree && removed.worktree() != null) {
// fleetd #283: this is the one cleanup step in this method that used to be bare. By the
// time it runs, the registry entry, the retained handle, and the pane are all already
@@ -404,6 +429,24 @@ public final class SessionManager implements TurnListener {
}
}
/**
* fleetd #316: the read that actually authorises {@code worktrees.remove}, taken with the
* worker's pane already stopped. Fails toward preserving (returns {@code true}) on any
* exception — the same rule the pre-stop check applies (CB-581): once we can no longer tell
* whether the worktree is dirty, preserving costs disk while deleting on a guess can destroy
* work that has no other copy.
*/
private boolean dirtyImmediatelyBeforeRemoval(MemberSession removed) {
try {
return worktrees.hasUncommitted(removed.worktree());
} catch (RuntimeException e) {
log.warn("release could not re-check worktree {} for pane={} terminal={} immediately "
+ "before removal; preserving it rather than risk destroying unsaved work: {}",
removed.worktree(), removed.paneId(), removed.terminalId(), e.toString());
return true;
}
}
/**
* Best-effort snapshot of a dirty worktree into {@code refs/wip/<branch>} (CB-578 stage C). A
* failure here must never escalate: the caller has already decided to preserve the worktree
@@ -0,0 +1,202 @@
package dev.ltms.fleet.config;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Constructor;
import java.lang.reflect.RecordComponent;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* fleetd #323: {@code ConfigRef.sameLaunchSettings} javadoc used to claim it "compares every
* component the launcher reads at spawn". It did not — {@code ideProjectDir}, {@code
* ideOpenCommand} and {@code autoCompactWindow} were all baked in at daemon startup (see
* {@code ClaudeCodeLauncher}/{@code OpenCodeLauncher}) and missing from the comparison, so a reload
* that changed only one of them reported "config reloaded" and the running daemon kept the old
* value.
*
* <p>This class is the mechanism the issue asked for: it enumerates every record component of
* {@link FleetConfig.Profile} by reflection and proves — by actually mutating a base profile one
* field at a time and calling the real method — that each component is either compared by
* {@link ConfigRef#sameLaunchSettings} or named in {@link ConfigRef#LAUNCH_SETTINGS_EXCLUDED} with
* a reason. A new profile field that is neither fails this test by name, not a hand-maintained list
* going stale.
*/
class ConfigRefProfileCoverageTest {
private static final RecordComponent[] COMPONENTS = FleetConfig.Profile.class.getRecordComponents();
/**
* One valid, non-blank value per record component — "the a value". None of these trip any
* defaulting/normalization in {@code Profile}'s compact constructor (see {@code
* FleetConfig.java}), so what goes in is what {@code sameLaunchSettings} sees back out.
*/
private static final Map<String, Object> BASE = baseValues();
/** The same shape, each value distinct from {@link #BASE} — "the b value". */
private static final Map<String, Object> ALT = altValues();
private static Map<String, Object> baseValues() {
Map<String, Object> v = new LinkedHashMap<>();
v.put("profile", "sonnet");
v.put("baseUrl", "http://gx00.gw:8000");
v.put("model", "sonnet");
v.put("configDir", "/config/a");
v.put("tokenEnv", "TOKEN_A");
v.put("argv", List.of("claude", "--flag-a"));
v.put("placement", "tab");
v.put("workspace", "workspace-a");
v.put("tabLabel", "label-a");
v.put("mcpUrl", "http://mcp-a");
v.put("cwd", "/cwd/a");
v.put("parityOverlay", List.of(".env", ".env.a"));
v.put("gitTokenEnv", "GIT_TOKEN_A");
v.put("gitHostEnv", "GITEA_HOST_A");
v.put("kind", "claude-code");
v.put("env", Map.of("K", "A"));
v.put("weight", 1.0f);
v.put("maxLoad", 5);
v.put("subscription", Boolean.TRUE);
v.put("exhaustedPattern", "usage limit a");
v.put("credentialId", "cred-a");
v.put("ideMcpUrl", "http://ide-mcp-a");
v.put("ideProjectDir", "modules/a");
v.put("ideOpenCommand", "open-cmd-a {dir}");
v.put("autoCompactWindow", 150000);
v.put("errorPattern", "error a");
assertNamesMatchComponents(v);
return v;
}
private static Map<String, Object> altValues() {
Map<String, Object> v = new LinkedHashMap<>();
v.put("profile", "sonnet-b");
v.put("baseUrl", "http://gx01.gw:8000");
v.put("model", "haiku");
v.put("configDir", "/config/b");
v.put("tokenEnv", "TOKEN_B");
v.put("argv", List.of("claude", "--flag-b"));
v.put("placement", "weighted");
v.put("workspace", "workspace-b");
v.put("tabLabel", "label-b");
v.put("mcpUrl", "http://mcp-b");
v.put("cwd", "/cwd/b");
v.put("parityOverlay", List.of(".env", ".env.b"));
v.put("gitTokenEnv", "GIT_TOKEN_B");
v.put("gitHostEnv", "GITEA_HOST_B");
v.put("kind", "opencode");
v.put("env", Map.of("K", "B"));
v.put("weight", 2.0f);
v.put("maxLoad", 9);
v.put("subscription", Boolean.FALSE);
v.put("exhaustedPattern", "usage limit b");
v.put("credentialId", "cred-b");
v.put("ideMcpUrl", "http://ide-mcp-b");
v.put("ideProjectDir", "modules/b");
v.put("ideOpenCommand", "open-cmd-b {dir}");
v.put("autoCompactWindow", 250000);
v.put("errorPattern", "error b");
assertNamesMatchComponents(v);
return v;
}
private static void assertNamesMatchComponents(Map<String, Object> values) {
Set<String> componentNames = new TreeSet<>();
for (RecordComponent rc : COMPONENTS) {
componentNames.add(rc.getName());
}
assertEquals(componentNames, new TreeSet<>(values.keySet()),
"this test's value map has drifted from FleetConfig.Profile's actual components — "
+ "update BASE/ALT alongside the record");
}
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);
}
/** {@code BASE} with exactly one named component swapped for its {@code ALT} value. */
private static FleetConfig.Profile mutate(String componentName) throws ReflectiveOperationException {
Map<String, Object> values = new LinkedHashMap<>(BASE);
values.put(componentName, ALT.get(componentName));
return profileOf(values);
}
/**
* The mechanism fleetd #323 asked for: enumerate {@link FleetConfig.Profile}'s record
* components, mutate each non-excluded one, and prove {@code sameLaunchSettings} actually
* notices — not just that some hand-maintained list claims it does. Prints the denominator
* (total / compared / excluded) the issue required: a checker that cannot state its own
* denominator is the failure this repo keeps hitting.
*/
@Test
void sameLaunchSettingsComparesEveryProfileComponentOrExcludesIt() throws ReflectiveOperationException {
int total = COMPONENTS.length;
Set<String> excluded = ConfigRef.LAUNCH_SETTINGS_EXCLUDED;
Set<String> allNames = new TreeSet<>();
for (RecordComponent rc : COMPONENTS) {
allNames.add(rc.getName());
}
assertTrue(allNames.containsAll(excluded),
"ConfigRef.LAUNCH_SETTINGS_EXCLUDED names a component that does not exist on "
+ "FleetConfig.Profile — check for a typo: " + excluded);
// The exclusion set is this mechanism's own escape hatch, so it has to be pinned too.
// Found by mutation while verifying fleetd #323: moving autoCompactWindow and
// ideOpenCommand OUT of the comparison and INTO the exclusion set left the whole suite
// green — the loop below simply skips them, and the denominator assertion still balances.
// That is exactly the lazy move a failing coverage test invites, and it silently restores
// the #323 bug. Only ideProjectDir and worktreeGroup were saved by a behavioural test in
// ConfigRefTest; the other two had none. So: growing this set now requires editing this
// line as well, which is a visible, deliberate diff rather than a quiet one.
assertEquals(Set.of("weight", "maxLoad", "credentialId"), excluded,
"ConfigRef.LAUNCH_SETTINGS_EXCLUDED changed. A component belongs in it ONLY if it "
+ "is read live off the config supplier, not baked into a launcher at "
+ "startup. If you are adding one to silence this test, that is fleetd #323 "
+ "happening again: compare it in sameLaunchSettings instead. If it really "
+ "is read live, name where it is read and update this assertion.");
FleetConfig.Profile base = profileOf(BASE);
List<String> uncovered = new java.util.ArrayList<>();
int compared = 0;
for (RecordComponent rc : COMPONENTS) {
String name = rc.getName();
if (excluded.contains(name)) {
continue;
}
FleetConfig.Profile mutated = mutate(name);
if (ConfigRef.sameLaunchSettings(base, mutated)) {
uncovered.add(name);
} else {
compared++;
}
}
System.out.printf(
"ConfigRef.sameLaunchSettings coverage — %d Profile components total, %d compared, "
+ "%d excluded (%s)%n",
total, compared, excluded.size(), excluded);
assertEquals(List.of(), uncovered,
"these FleetConfig.Profile components changed but ConfigRef.sameLaunchSettings "
+ "reported no difference — add each one to the comparison (it is read at "
+ "spawn and baked in until a restart) or to ConfigRef.LAUNCH_SETTINGS_EXCLUDED "
+ "with a reason it is genuinely read live: " + uncovered);
assertEquals(total, compared + excluded.size(),
"every FleetConfig.Profile record component must be either compared or excluded — "
+ total + " components, " + compared + " compared, " + excluded.size()
+ " excluded");
}
}
@@ -434,6 +434,364 @@ class ConfigRefTest {
assertEquals("provider 5xx", ref.get().profiles().get("sonnet").errorPattern());
}
/**
* fleetd #323 instance 1: {@code ideProjectDir} is read at spawn off the frozen profile map
* (see {@code ClaudeCodeLauncher}/{@code OpenCodeLauncher}) exactly like {@code model}, but was
* missing from {@code sameLaunchSettings} — a reload changing only this field used to report a
* bare "config reloaded" and the running daemon kept launching with the old value.
*/
@Test
void changingAProfilesIdeProjectDirIsReportedAsDeferred(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, """
bind:
host: 127.0.0.1
port: 8765
herdrSocket: ~/.config/herdr/herdr.sock
profiles:
sonnet:
baseUrl: http://gx00.gw:8000
model: sonnet
ideProjectDir: fleetd
guard:
offSubscriptionHosts:
- gx00.gw
""");
ConfigRef ref = refFor(f);
Files.writeString(f, """
bind:
host: 127.0.0.1
port: 8765
herdrSocket: ~/.config/herdr/herdr.sock
profiles:
sonnet:
baseUrl: http://gx00.gw:8000
model: sonnet
ideProjectDir: fleetd-renamed
guard:
offSubscriptionHosts:
- gx00.gw
""");
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(1, out.deferred().size(), out.deferred().toString());
assertTrue(out.deferred().getFirst().contains("sonnet"), out.deferred().toString());
assertTrue(out.deferred().getFirst().contains("launch settings"), out.deferred().toString());
// The snapshot still carries the new value — a restart is what makes it take effect.
assertEquals("fleetd-renamed", ref.get().profiles().get("sonnet").ideProjectDir());
}
/**
* fleetd #323 instance 2: {@code worktreeGroup} is baked into the same {@code GitWorktrees}
* as {@code worktreeRoot} (Fleetd.java:251) and never rebuilt, but only {@code worktreeRoot}
* was on {@code changedDeferredKeys} — a reload changing only the group reported a bare
* "config reloaded" and newly provisioned worktrees kept the old sharing behaviour.
*/
@Test
void changingWorktreeGroupIsReportedAsDeferred(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("worktreeGroup: devgroup\n"));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("worktreeGroup: devgroup2\n"));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(java.util.List.of("worktreeGroup"), out.deferred());
assertTrue(out.summary().contains("needs a restart") || out.summary().contains("need a restart"),
out.summary());
// The snapshot still carries the new value — a restart is what makes it take effect.
assertEquals("devgroup2", ref.get().worktreeGroup());
}
/**
* fleetd #326: {@code primary} is read only off the startup snapshot — {@code Fleetd.java:506,
* 519, 520} feed {@code PrimaryRegistry} and {@code ReplyPushLoop} at construction and neither is
* rebuilt on reload — but it was missing from {@link ConfigRef#changedDeferredKeys}, so a reload
* that only changed the pinned primary terminal reported a bare "config reloaded" while a lead
* whose tab no longer matched stayed demoted to worker.
*/
@Test
void changingPrimaryIsReportedAsDeferred(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
primary:
terminal: term-a
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
primary:
terminal: term-b
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(java.util.List.of("primary"), out.deferred());
assertTrue(out.summary().contains("need") && out.summary().contains("restart"), out.summary());
// The snapshot still carries the new value — a restart is what makes it take effect.
assertEquals("term-b", ref.get().primary().terminal());
}
/**
* fleetd #326: {@code configReload} itself is read only at startup ({@code Fleetd.java:679-680})
* to decide whether to build a {@code ConfigWatcher} at all, and with what interval — the watcher
* that would apply a later change is itself built once, so it is deferred rather than cold (see
* {@link ConfigRef}'s class doc: cold means an already-open resource would go inconsistent with
* the new value, and there is no such resource here — a running watcher just keeps polling on its
* original enabled/interval until a restart, exactly like {@code lifecycle} or {@code guard}).
* Before this fix, turning reload off (or changing its interval) through a reload reported a bare
* "config reloaded" — the obvious joke the issue names.
*/
@Test
void changingConfigReloadIsReportedAsDeferred(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
configReload:
enabled: true
intervalSeconds: 10
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
configReload:
enabled: false
intervalSeconds: 30
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(java.util.List.of("configReload"), out.deferred());
assertTrue(out.summary().contains("need") && out.summary().contains("restart"), out.summary());
// The snapshot still carries the new value — a restart is what makes it take effect.
assertFalse(ref.get().configReload().isEnabled());
assertEquals(30, ref.get().configReload().intervalSeconds());
}
/**
* fleetd #330: {@code health:} is read both ways — {@code Fleetd.java:556-563} builds the
* monitor off the startup snapshot and never rebuilds it, but {@code Fleetd.java:648-650} reads
* {@code config.get().health()} live on every {@code fleet_profiles} call. A changed value is
* neither purely hot nor purely deferred, so it gets its own {@code split} report naming both
* halves rather than a bare "config reloaded" (which would hide the frozen half) or a plain
* {@code deferred} entry (which would hide that the coverage string already applied).
*/
@Test
void changingHealthIsReportedAsSplit(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
health:
enabled: true
intervalSeconds: 30
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
health:
enabled: true
intervalSeconds: 90
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertTrue(out.deferred().isEmpty(), out.deferred().toString());
assertEquals(1, out.split().size(), out.split().toString());
assertTrue(out.split().getFirst().startsWith("health:"), out.split().toString());
assertTrue(out.split().getFirst().contains("restart"), out.split().toString());
assertTrue(out.split().getFirst().contains("live"), out.split().toString());
assertTrue(out.summary().contains("partially live"), out.summary());
// The snapshot still carries the new value — the monitor itself is what waits for a restart.
assertEquals(90, ref.get().health().intervalSeconds());
}
/**
* fleetd #330: {@code coordinator:} is the other split key — {@code Fleetd.java:502} opens the
* {@code LeadMailbox} off the startup snapshot and never reopens it, but
* {@code HerdrPeerLauncher.java:1530} reads {@code config.get().coordinator()} live on every
* spawn to keep the broker URI env-var name out of a member's environment.
*/
@Test
void changingCoordinatorIsReportedAsSplit(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
coordinator:
selfId: mac-a
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
coordinator:
selfId: mac-b
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertTrue(out.deferred().isEmpty(), out.deferred().toString());
assertEquals(1, out.split().size(), out.split().toString());
assertTrue(out.split().getFirst().startsWith("coordinator:"), out.split().toString());
assertTrue(out.split().getFirst().contains("restart"), out.split().toString());
assertTrue(out.split().getFirst().contains("live"), out.split().toString());
// The snapshot still carries the new value — the LeadMailbox connection is what waits for a
// restart; selfId names this daemon's own inbox queue and a peer cannot discover a rename.
assertEquals("mac-b", ref.get().coordinator().selfId());
}
/**
* fleetd #333: {@code fleet:} is split too — {@code fleet.leaders} is read only at
* {@code Fleetd.java:281} to build the {@code LeadTabScanner}'s identity map (fed into
* {@code CallerResolver}) and to auto-launch leads via {@code LeadLauncher.ensureLeads()}, and
* neither is rebuilt on reload, while the rest of {@code fleet:} (role pools, charters,
* tabLabel) is read live through the {@code CompositePeerLauncher} supplier. Before this fix
* {@code fleet} sat in the coverage checker's hot-excluded escape hatch, so a reload that
* changed only {@code fleet.leaders} reported a bare "config reloaded" — exactly the
* under-claim the split class exists to prevent for {@code health}/{@code coordinator}.
*/
@Test
void changingFleetLeadersIsReportedAsSplit(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
fleet:
leaders:
opus:
tab: "lead: opus-a"
profile: sonnet
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
fleet:
leaders:
opus:
tab: "lead: opus-b"
profile: sonnet
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertTrue(out.deferred().isEmpty(), out.deferred().toString());
assertEquals(1, out.split().size(), out.split().toString());
assertTrue(out.split().getFirst().startsWith("fleet:"), out.split().toString());
assertTrue(out.split().getFirst().contains("restart"), out.split().toString());
assertTrue(out.split().getFirst().contains("live"), out.split().toString());
assertTrue(out.summary().contains("partially live"), out.summary());
// The snapshot still carries the new value — the LeadTabScanner's identity map and
// LeadLauncher's auto-launch are what wait for a restart; a reload rebuilds neither.
assertEquals("lead: opus-b", ref.get().fleet().leaders().get("opus").tab());
}
/**
* fleetd #333: {@code fleet.leaders} is the ONLY frozen part of {@code fleet:}. A reload that
* changes {@code tabLabel} (or charters, or a role pool) without touching {@code fleet.leaders}
* must stay fully hot with nothing reported — proving {@link ConfigRef#changedSplitKeys}
* compares {@code fleet.leaders} specifically rather than the whole {@code Fleet} record, which
* would over-claim "needs a restart" for a change that is genuinely all live (the mirror
* mistake of the under-claim this class exists to prevent).
*/
@Test
void changingFleetTabLabelWithoutLeadersStaysFullyHot(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
fleet:
tabLabel: "{role}: {profile} #{n}"
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
fleet:
tabLabel: "[{profile}] {role}"
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertTrue(out.deferred().isEmpty(), out.deferred().toString());
assertTrue(out.split().isEmpty(), out.split().toString());
assertEquals("config reloaded", out.summary());
assertEquals("[{profile}] {role}", ref.get().fleet().tabLabel());
}
/**
* A split change must not refuse the reload (invariant 2 of fleetd #330) and
* {@code Outcome.applied()} must stay {@code true} (invariant 3) — unlike a cold change, the
* live half of a split key genuinely took effect, so refusing would throw that away.
*/
@Test
void aSplitChangeDoesNotRefuseTheReload(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("coordinator:\n selfId: mac-a\n"));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("coordinator:\n selfId: mac-b\n"));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertTrue(out.error() == null);
assertTrue(out.coldKeys().isEmpty());
}
/**
* Both split keys can change in one reload — the report names both, and a caller reading
* {@code split} does not have to guess which half of which key already applied.
*/
@Test
void changingBothSplitKeysReportsBoth(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
health:
enabled: true
coordinator:
selfId: mac-a
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
health:
enabled: false
coordinator:
selfId: mac-b
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(2, out.split().size(), out.split().toString());
assertTrue(out.split().stream().anyMatch(s -> s.startsWith("health:")), out.split().toString());
assertTrue(out.split().stream().anyMatch(s -> s.startsWith("coordinator:")), out.split().toString());
}
/**
* A split key and a deferred key changing in the same reload must both show up, each in its own
* list — proving the two fields do not step on each other and {@link ConfigRef.Outcome#summary()}
* reports both halves of the message.
*/
@Test
void aSplitChangeAndADeferredChangeCoexist(@TempDir Path dir) throws Exception {
Path f = dir.resolve("fleetd.yaml");
Files.writeString(f, yaml("""
coordinator:
selfId: mac-a
lifecycle:
drainTimeoutSeconds: 30
"""));
ConfigRef ref = refFor(f);
Files.writeString(f, yaml("""
coordinator:
selfId: mac-b
lifecycle:
drainTimeoutSeconds: 60
"""));
ConfigRef.Outcome out = ref.reload();
assertTrue(out.applied());
assertEquals(java.util.List.of("lifecycle"), out.deferred());
assertEquals(1, out.split().size(), out.split().toString());
assertTrue(out.split().getFirst().startsWith("coordinator:"), out.split().toString());
assertTrue(out.summary().contains("need a restart") || out.summary().contains("needs a restart"),
out.summary());
assertTrue(out.summary().contains("partially live"), out.summary());
}
@Test
void aFixedRefHasNoFileAndRefusesToReload() {
FleetConfig cfg = new FleetConfig(null, null, null, null, null, null,
@@ -0,0 +1,171 @@
package dev.ltms.fleet.config;
import org.junit.jupiter.api.Test;
import java.lang.reflect.RecordComponent;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* fleetd #330: a new top-level {@link FleetConfig} record component must be triaged into a reload
* class before it ships, or it repeats fleetd #323 ({@code worktreeGroup} missing from {@code
* ConfigRef.changedDeferredKeys}) and fleetd #326 ({@code primary}/{@code configReload} missing the
* same way) — a key silently absent from {@link ConfigRef}'s reload machinery, so a reload changing
* only that key reports a bare "config reloaded" for a change the running daemon never picked up.
*
* <p>This is the top-level counterpart of {@link ConfigRefProfileCoverageTest}: instead of
* enumerating {@link FleetConfig.Profile}'s record components, it enumerates {@link FleetConfig}'s
* own — {@code bind}, {@code health}, {@code memberCredentials}, and so on — and requires each to
* fall into exactly one of four homes: {@link ConfigRef#COLD_KEYS}, {@link #DEFERRED_TOP_LEVEL_KEYS}
* (compared in {@code ConfigRef.changedDeferredKeys}), {@link ConfigRef#SPLIT_KEYS}, or
* {@link #HOT_EXCLUDED_TOP_LEVEL_KEYS} (the escape hatch: read live off the config supplier, so no
* reload bookkeeping is needed for it at all).
*
* <h2>What this checker can and cannot prove</h2>
* It proves the record's <em>shape</em> is fully triaged: every one of {@code FleetConfig}'s
* components sits in exactly one of the four sets, none sits in two, and the escape hatch
* ({@link #HOT_EXCLUDED_TOP_LEVEL_KEYS}) cannot silently grow without a visible diff to this file.
* That is what "a new component cannot be added without someone triaging it" means in practice.
*
* <p>It CANNOT prove that any of the citations are <em>true</em>. "Compared in {@code
* changedDeferredKeys}" and "read live off {@code config.get()}" are facts about {@code
* ConfigRef.java}, {@code Fleetd.java} and {@code HerdrPeerLauncher.java} that a reflection-only
* test over {@code FleetConfig}'s shape has no way to inspect — this test would pass identically
* whether or not the cited line still does what the comment next to it says. Trust the citation
* because a person read the source (the exact call sites are named next to each set below), not
* because this test is green. {@link ConfigRefTest} is what behaviourally proves the deferred and
* split keys it covers actually get reported; {@link ConfigRefProfileCoverageTest} does the same,
* behaviourally, for {@code FleetConfig.Profile}'s own fields.
*/
class ConfigRefTopLevelCoverageTest {
private static final RecordComponent[] COMPONENTS = FleetConfig.class.getRecordComponents();
/**
* Top-level components whose change {@code ConfigRef.changedDeferredKeys} reads and reports on.
* Fleetd #337 promoted this out of a hand-maintained copy here into {@link ConfigRef#DEFERRED_KEYS}
* itself, the same reason {@link ConfigRef#COLD_KEYS} and {@link ConfigRef#SPLIT_KEYS} are
* production constants rather than test-side copies: two lists that are supposed to describe the
* same method are exactly the shape that silently drifts apart. {@code spawnReadyTimeoutMs}/
* {@code spawnReadyPollMs} are compared together and reported under one combined label
* ({@code "spawnReady*"}); {@code profiles} is compared twice over — once for added/removed
* profile names, once for an existing profile's launch settings — and that second comparison
* excludes {@code weight}/{@code maxLoad}/{@code credentialId} as hot sub-fields, which is what
* {@link ConfigRefProfileCoverageTest} exists to keep honest at the sub-field level. {@code
* profiles} itself still belongs here, not in the hot-exclusion set below: most of a profile's
* fields are NOT read live, so citing "read live off the config supplier" for the whole
* top-level key would be false.
*/
private static final Set<String> DEFERRED_TOP_LEVEL_KEYS = ConfigRef.DEFERRED_KEYS;
/**
* The escape hatch: top-level components with no reload bookkeeping at all, because every read
* of them goes live through {@link ConfigRef#get()} rather than off a startup snapshot. A
* component belongs here ONLY if that is true — never because adding it here makes this test
* pass. fleetd #323 is the cautionary tale for exactly this pattern: the identical hatch on
* {@code ConfigRef.LAUNCH_SETTINGS_EXCLUDED} let two profile fields be silently re-broken with
* the whole suite green, and it was only caught by mutating the checker itself (see this class's
* own mutation test below, and {@code ConfigRefProfileCoverageTest}'s equivalent).
*
* <ul>
* <li>{@code placement} — read live by the placement policy on every spawn (class doc, Hot
* bullet; {@code ConfigRefTest.aConsumerHoldingTheRefSeesTheNewValue} proves it
* behaviourally).</li>
* <li>{@code memberCredentials} — read live at {@code Fleetd.java:198, 205, 729}.</li>
* <li>{@code memberLoginShell} — read live at
* {@code HerdrPeerLauncher#configuredMemberLoginShell}.</li>
* </ul>
*
* <p>{@code fleet} used to sit here too, on the strength of most of it (role pools, charters,
* tabLabel) being read the same live way — but {@code fleet.leaders} inside the same key is
* read only at startup and genuinely needs a restart, which is a real gap this escape hatch
* cannot represent: it excuses a whole top-level key from reload bookkeeping, and {@code fleet}
* needed exactly half of it excused. fleetd #333 moved it into {@link ConfigRef#SPLIT_KEYS}
* instead, where {@code changedSplitKeys} reports the frozen half by name. That is the
* cautionary tale this test's own javadoc already told: this checker proves the record's shape
* is triaged, never that a bucket a key sits in is the right one — a person has to read the
* source, which is exactly how fleetd #333 found {@code fleet} sitting in the wrong bucket
* while this test stayed green throughout.</p>
*/
private static final Set<String> HOT_EXCLUDED_TOP_LEVEL_KEYS =
Set.of("placement", "memberCredentials", "memberLoginShell");
@Test
void everyTopLevelComponentIsAccountedForInExactlyOneClass() {
Set<String> allNames = new TreeSet<>();
for (RecordComponent rc : COMPONENTS) {
allNames.add(rc.getName());
}
Set<String> cold = ConfigRef.COLD_KEYS;
Set<String> split = ConfigRef.SPLIT_KEYS;
Set<String> deferred = DEFERRED_TOP_LEVEL_KEYS;
Set<String> hot = HOT_EXCLUDED_TOP_LEVEL_KEYS;
// Typo guard on each set — the same check ConfigRefProfileCoverageTest runs on
// LAUNCH_SETTINGS_EXCLUDED. A name that does not exist on FleetConfig is a silent no-op.
assertTrue(allNames.containsAll(cold),
"ConfigRef.COLD_KEYS names a component that does not exist on FleetConfig: " + cold);
assertTrue(allNames.containsAll(split),
"ConfigRef.SPLIT_KEYS names a component that does not exist on FleetConfig: " + split);
assertTrue(allNames.containsAll(deferred),
"DEFERRED_TOP_LEVEL_KEYS names a component that does not exist on FleetConfig: " + deferred);
assertTrue(allNames.containsAll(hot),
"HOT_EXCLUDED_TOP_LEVEL_KEYS names a component that does not exist on FleetConfig: " + hot);
// The escape hatch is pinned. Growing it requires editing this line — a visible, deliberate
// diff, not a quiet one. See the field javadoc above for what "belongs here" actually means.
assertEquals(Set.of("placement", "memberCredentials", "memberLoginShell"), hot,
"HOT_EXCLUDED_TOP_LEVEL_KEYS changed. A component belongs here ONLY if it is read "
+ "live off the config supplier, never because adding it makes this test "
+ "pass. If you are adding one to silence this test, that is fleetd #323 "
+ "happening again: account for it in ConfigRef.changedDeferredKeys (or "
+ "COLD_KEYS/SPLIT_KEYS) instead. If it really is read live, name where and "
+ "update this assertion and the field javadoc together.");
// No component may sit in two buckets at once — the denominator check below could not catch
// that on its own (two buckets double-booking one key still sums to the right total if
// another key is simultaneously missing), so check every pair directly and name the culprit.
record Bucket(String name, Set<String> keys) {}
List<Bucket> buckets = List.of(
new Bucket("COLD_KEYS", cold), new Bucket("SPLIT_KEYS", split),
new Bucket("DEFERRED_TOP_LEVEL_KEYS", deferred), new Bucket("HOT_EXCLUDED_TOP_LEVEL_KEYS", hot));
for (int i = 0; i < buckets.size(); i++) {
for (int j = i + 1; j < buckets.size(); j++) {
Set<String> overlap = new LinkedHashSet<>(buckets.get(i).keys());
overlap.retainAll(buckets.get(j).keys());
assertEquals(Set.of(), overlap, "a component is in both " + buckets.get(i).name()
+ " and " + buckets.get(j).name() + ": " + overlap);
}
}
Set<String> union = new TreeSet<>();
union.addAll(cold);
union.addAll(split);
union.addAll(deferred);
union.addAll(hot);
System.out.printf(
"FleetConfig top-level coverage — %d components total: %d cold %s, %d deferred %s, "
+ "%d split %s, %d hot-excluded %s%n",
allNames.size(), cold.size(), cold, deferred.size(), deferred, split.size(), split,
hot.size(), hot);
Set<String> missing = new TreeSet<>(allNames);
missing.removeAll(union);
assertEquals(Set.of(), missing,
"these FleetConfig components are in none of COLD_KEYS, DEFERRED_TOP_LEVEL_KEYS, "
+ "SPLIT_KEYS or HOT_EXCLUDED_TOP_LEVEL_KEYS — triage each one into whichever "
+ "actually describes it: " + missing);
assertEquals(allNames.size(), cold.size() + deferred.size() + split.size() + hot.size(),
"counts don't sum to the component total even though every component was found in "
+ "the union — " + allNames.size() + " components, " + cold.size()
+ " cold + " + deferred.size() + " deferred + " + split.size() + " split + "
+ hot.size() + " hot-excluded");
}
}
@@ -0,0 +1,284 @@
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.Map;
import java.util.Set;
import java.util.TreeSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* fleetd #333, finding F2: {@link ConfigRefTopLevelCoverageTest} proves every {@link FleetConfig}
* top-level component sits in exactly one of {@link ConfigRef#COLD_KEYS}, {@link
* ConfigRef#DEFERRED_KEYS}, {@link ConfigRef#SPLIT_KEYS} or the hot-excluded set. It does
* <strong>not</strong> prove that a key's membership in one of the first three sets corresponds to
* any actual comparison in {@link ConfigRef}: a key can sit in a set with no branch in {@code
* changedColdKeys}/{@code changedSplitKeys}/{@code changedDeferredKeys} checking it, and both
* {@link ConfigRefTopLevelCoverageTest} and the "kept in step" {@code assert} inside the first two
* of those methods stay green, because neither one reads the method body — the coverage test only
* reads set membership, and the assert only checks that reported entries are a SUBSET of the set,
* never that every set member produced a reported entry. {@code changedDeferredKeys} does not even
* have a "kept in step" assert of its own.
*
* <p>Measured directly, live, while fixing fleetd #333: dropping the {@code coordinator} branch out
* of {@code ConfigRef.changedSplitKeys} while leaving {@code "coordinator"} in
* {@link ConfigRef#SPLIT_KEYS} left {@link ConfigRefTopLevelCoverageTest} and the in-method assert
* both green — only a hand-written behavioural case in {@link ConfigRefTest} caught it, because it
* happened to name that exact key. This is the {@link ConfigRefProfileCoverageTest} mechanism one
* level up, generalised over every {@code COLD_KEYS}/{@code SPLIT_KEYS}/{@code DEFERRED_KEYS}
* member rather than one hand-picked field: enumerate {@link FleetConfig}'s own record components by
* reflection, build "a config where only {@code <key>} differs" for each key, and call the real
* {@link ConfigRef#changedColdKeys}/{@link ConfigRef#changedSplitKeys}/
* {@link ConfigRef#changedDeferredKeys} methods (all package-private for exactly this, the same
* reason {@link ConfigRef#sameLaunchSettings} already is) to prove each one is actually reported —
* not assumed from a set literal.
*
* <h2>fleetd #337 — DEFERRED_KEYS was the gap left open here</h2>
* This class originally covered {@code COLD_KEYS} and {@code SPLIT_KEYS} only — {@code
* DEFERRED_TOP_LEVEL_KEYS} (now {@link ConfigRef#DEFERRED_KEYS}) carried the identical one-way risk
* in principle, unexercised. fleetd #337 measured the real consequence rather than assuming it from
* the shape of the gap: dropping {@code guard}'s comparison out of {@code changedDeferredKeys} while
* {@code "guard"} stayed in the set left all 1355 tests green — the same failure mode {@code
* coordinator} demonstrated for {@code SPLIT_KEYS} in fleetd #333, now confirmed for {@code
* DEFERRED_KEYS} too. Re-deriving the full list by mutation (drop each key's branch in turn, run the
* suite, restore) found six of the eleven {@code DEFERRED_KEYS} members with no behavioural test in
* {@link ConfigRefTest} naming them: {@code guard}, {@code leadHeartbeat}, {@code worktreeRoot},
* {@code spawnReadyTimeoutMs}, {@code spawnReadyPollMs} and {@code quarantineCooldownSeconds}. That
* list corrects fleetd #333's own guess at it in two ways the mutation proved and a reading did not:
* {@code lifecycle} is NOT on it — {@code ConfigRefTest.aDeferredChangeIsAppliedAndReported} already
* names it, and dropping its branch fails that test — and {@code worktreeRoot} IS on it, which #333
* never named at all. The other five {@code DEFERRED_KEYS} members ({@code lifecycle}, {@code
* worktreeGroup}, {@code primary}, {@code configReload}, {@code profiles}) already had a hand-written
* case each. {@link #everyDeferredKeyIsActuallyReportedByChangedDeferredKeys} below now covers all
* eleven the reflective way, so the six with no hand-written test are no longer silently unpinned —
* every {@code DEFERRED_KEYS} component turned out to be a scalar or a simple record, so, unlike
* {@code fleet.leaders} in fleetd #333, none needed an exclusion set: {@link #BASE}/{@link #ALT} give
* every top-level component (not only {@code COLD_KEYS}/{@code SPLIT_KEYS}) a real, distinct value.
*/
class ConfigRefTopLevelReportingCoverageTest {
private static final RecordComponent[] COMPONENTS = FleetConfig.class.getRecordComponents();
/**
* One valid value per top-level {@link FleetConfig} component — "the a value". fleetd #337 gave
* every {@code DEFERRED_KEYS} component a real value here too (previously left {@code null} on
* both sides, which meant {@code mutate(key)} produced no actual difference for any of them);
* only {@code placement}, {@code memberCredentials} and {@code memberLoginShell} — the
* hot-excluded set, never compared by any {@code changed*Keys} method — stay {@code null}.
* {@link FleetConfig}'s compact constructor only normalizes {@code profiles}, so every other
* field accepts whatever is put here unmutated.
*/
private static final Map<String, Object> BASE = baseValues();
/** The same shape, each value distinct from {@link #BASE} — "the b value". */
private static final Map<String, Object> ALT = altValues();
private static Map<String, Object> baseValues() {
Map<String, Object> v = new LinkedHashMap<>();
v.put("bind", new FleetConfig.Bind("127.0.0.1", 8765));
v.put("herdrSocket", "~/.config/herdr/a.sock");
v.put("memberHerdrSocket", "~/.config/herdr/member-a.sock");
v.put("profiles", Map.of());
v.put("guard", new FleetConfig.Guard(List.of("host-a")));
v.put("worktreeRoot", "/wt/a");
v.put("lifecycle", new FleetConfig.Lifecycle(300, 5, 30, false));
v.put("spawnReadyTimeoutMs", 5000);
v.put("spawnReadyPollMs", 100);
v.put("broker", new FleetConfig.Broker("amqp://a", null, 1));
v.put("primary", new FleetConfig.Primary("term-a", 1, 1000));
v.put("fleet", new FleetConfig.Fleet(
Map.of("opus", new FleetConfig.Leader("sonnet", "lead: opus-a", 1, null, 10,
"claude", null, null, null)),
Map.of(), Map.of(), Map.of(), Map.of(), "{role}: {profile} #{n}"));
v.put("leadHeartbeat", new FleetConfig.LeadHeartbeat(300, 60_000L, 3));
v.put("health", new FleetConfig.Health(true, 30, 600, null, null));
v.put("placement", null);
v.put("auth", new FleetConfig.Auth("loopback-trust", null));
v.put("configReload", new FleetConfig.ConfigReload(true, 10));
v.put("quarantineCooldownSeconds", 1800);
v.put("memberCredentials", null);
v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-a", null, "self-a", 1));
v.put("worktreeGroup", "group-a");
v.put("memberLoginShell", null);
assertNamesMatchComponents(v);
return v;
}
private static Map<String, Object> altValues() {
Map<String, Object> v = new LinkedHashMap<>();
v.put("bind", new FleetConfig.Bind("127.0.0.2", 8766));
v.put("herdrSocket", "~/.config/herdr/b.sock");
v.put("memberHerdrSocket", "~/.config/herdr/member-b.sock");
// A single added profile — enough to trip the "added/removed" comparison in
// ConfigRef.changedDeferredKeys, which is all this mechanism needs to prove "profiles" has
// a branch behind it; the launch-settings comparison already has its own hand-written cases
// in ConfigRefTest (changingAProfilesLaunchSettingsIsReportedAsDeferred and siblings).
v.put("profiles", Map.of("sonnet", minimalProfile("sonnet")));
v.put("guard", new FleetConfig.Guard(List.of("host-b")));
v.put("worktreeRoot", "/wt/b");
v.put("lifecycle", new FleetConfig.Lifecycle(600, 10, 60, true));
v.put("spawnReadyTimeoutMs", 10_000);
v.put("spawnReadyPollMs", 200);
v.put("broker", new FleetConfig.Broker("amqp://b", null, 2));
v.put("primary", new FleetConfig.Primary("term-b", 2, 2000));
// Differs from BASE.fleet only in fleet.leaders (a different tab for "opus") — the frozen
// sub-field ConfigRef.changedSplitKeys actually compares. A Fleet that instead differed only
// in tabLabel would correctly NOT be reported (see
// ConfigRefTest.changingFleetTabLabelWithoutLeadersStaysFullyHot) and would wrongly fail this
// test — that is by design, not a gap: this map exists to prove fleet.leaders is covered.
v.put("fleet", new FleetConfig.Fleet(
Map.of("opus", new FleetConfig.Leader("sonnet", "lead: opus-b", 1, null, 10,
"claude", null, null, null)),
Map.of(), Map.of(), Map.of(), Map.of(), "{role}: {profile} #{n}"));
v.put("leadHeartbeat", new FleetConfig.LeadHeartbeat(600, 120_000L, 5));
v.put("health", new FleetConfig.Health(false, 90, 900, null, null));
v.put("placement", null);
v.put("auth", new FleetConfig.Auth("token", "TOKEN_ENV"));
v.put("configReload", new FleetConfig.ConfigReload(false, 20));
v.put("quarantineCooldownSeconds", 3600);
v.put("memberCredentials", null);
v.put("coordinator", new FleetConfig.Coordinator("amqp://coord-b", null, "self-b", 2));
v.put("worktreeGroup", "group-b");
v.put("memberLoginShell", null);
assertNamesMatchComponents(v);
return v;
}
/** A minimal, otherwise-null {@link FleetConfig.Profile} — just enough to name one in a map. */
private static FleetConfig.Profile minimalProfile(String name) {
return new FleetConfig.Profile(name, null, null, null, null, null, null, null, null, null,
null, null, null, null, null, null, null, null, null, null, null, null, null, null,
null, null);
}
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's actual top-level components — "
+ "update BASE/ALT alongside the record");
}
private static FleetConfig configOf(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> ctor = FleetConfig.class.getDeclaredConstructor(types);
return ctor.newInstance(args);
}
/** {@code BASE} with exactly one named top-level component swapped for its {@code ALT} value. */
private static FleetConfig mutate(String componentName) throws ReflectiveOperationException {
Map<String, Object> values = new LinkedHashMap<>(BASE);
values.put(componentName, ALT.get(componentName));
return configOf(values);
}
/**
* The mechanism fleetd #333 F2 asked for, applied to {@link ConfigRef#COLD_KEYS}: mutate each
* cold key in isolation and prove {@code changedColdKeys} actually names it, not just that
* {@code COLD_KEYS} claims it does.
*/
@Test
void everyColdKeyIsActuallyReportedByChangedColdKeys() throws ReflectiveOperationException {
FleetConfig base = configOf(BASE);
List<String> uncovered = new ArrayList<>();
for (String key : new TreeSet<>(ConfigRef.COLD_KEYS)) {
FleetConfig mutated = mutate(key);
if (!ConfigRef.changedColdKeys(base, mutated).contains(key)) {
uncovered.add(key);
}
}
System.out.printf(
"ConfigRef.changedColdKeys reporting coverage — %d COLD_KEYS, %d verified%n",
ConfigRef.COLD_KEYS.size(), ConfigRef.COLD_KEYS.size() - uncovered.size());
assertEquals(List.of(), uncovered,
"these keys are in ConfigRef.COLD_KEYS but mutating them alone produces no matching "
+ "entry from changedColdKeys — a set entry with no comparison behind it: "
+ uncovered);
}
/**
* The same mechanism applied to {@link ConfigRef#SPLIT_KEYS}: mutate each split key in
* isolation and prove {@code changedSplitKeys} actually names it (message starting with
* {@code "<key>:"}), not just that {@code SPLIT_KEYS} claims it does. This is the exact check
* that would have failed fleetd #333's own reproduction — dropping the {@code coordinator}
* branch from {@code changedSplitKeys} while {@code "coordinator"} stayed in {@code SPLIT_KEYS}
* — see the mutation proof in the fleetd #333 PR description; the earlier checkers here (the
* shape test and the in-method assert) do not.
*/
@Test
void everySplitKeyIsActuallyReportedByChangedSplitKeys() throws ReflectiveOperationException {
FleetConfig base = configOf(BASE);
List<String> uncovered = new ArrayList<>();
for (String key : new TreeSet<>(ConfigRef.SPLIT_KEYS)) {
FleetConfig mutated = mutate(key);
List<String> split = ConfigRef.changedSplitKeys(base, mutated);
if (split.stream().noneMatch(s -> s.startsWith(key + ":"))) {
uncovered.add(key);
}
}
System.out.printf(
"ConfigRef.changedSplitKeys reporting coverage — %d SPLIT_KEYS, %d verified%n",
ConfigRef.SPLIT_KEYS.size(), ConfigRef.SPLIT_KEYS.size() - uncovered.size());
assertEquals(List.of(), uncovered,
"these keys are in ConfigRef.SPLIT_KEYS but mutating them alone produces no matching "
+ "entry from changedSplitKeys — a set entry with no comparison behind it, "
+ "exactly the fleetd #333 F2 shape: " + uncovered);
}
/**
* For most {@link ConfigRef#DEFERRED_KEYS} members, {@code changedDeferredKeys} reports the key
* name verbatim — the default this map assumes. Two entries don't: {@code spawnReadyTimeoutMs}
* and {@code spawnReadyPollMs} are compared together in one branch and reported under the
* combined label {@code "spawnReady*"} (see {@link ConfigRef#changedDeferredKeys}). {@code
* profiles} keeps the default: mutating it here only exercises the added/removed comparison
* (see {@link #altValues}), which reports {@code "profiles (added/removed: …)"} — starts with
* {@code "profiles"}, same as the default would expect.
*/
private static final Map<String, String> DEFERRED_REPORT_PREFIX = Map.of(
"spawnReadyTimeoutMs", "spawnReady*",
"spawnReadyPollMs", "spawnReady*");
/**
* fleetd #337: the same mechanism applied to {@link ConfigRef#DEFERRED_KEYS}, closing the gap
* this class's own javadoc left open since fleetd #333. Mutate each deferred key in isolation
* and prove {@code changedDeferredKeys} actually names it (message starting with the key's
* expected report prefix — see {@link #DEFERRED_REPORT_PREFIX}), not just that {@code
* DEFERRED_KEYS} claims it does. This is the exact check that fails for {@code guard} the way
* {@code coordinator} failed {@link #everySplitKeyIsActuallyReportedByChangedSplitKeys} in
* fleetd #333 — verified live: dropping {@code guard}'s branch from {@code changedDeferredKeys}
* while {@code "guard"} stayed in {@code DEFERRED_KEYS} left the whole 1355-test suite green,
* and this test is what now catches it (it fails naming {@code guard} with that mutation in
* place).
*/
@Test
void everyDeferredKeyIsActuallyReportedByChangedDeferredKeys() throws ReflectiveOperationException {
FleetConfig base = configOf(BASE);
List<String> uncovered = new ArrayList<>();
for (String key : new TreeSet<>(ConfigRef.DEFERRED_KEYS)) {
FleetConfig mutated = mutate(key);
List<String> deferred = ConfigRef.changedDeferredKeys(base, mutated);
String prefix = DEFERRED_REPORT_PREFIX.getOrDefault(key, key);
if (deferred.stream().noneMatch(s -> s.startsWith(prefix))) {
uncovered.add(key);
}
}
System.out.printf(
"ConfigRef.changedDeferredKeys reporting coverage — %d DEFERRED_KEYS, %d verified%n",
ConfigRef.DEFERRED_KEYS.size(), ConfigRef.DEFERRED_KEYS.size() - uncovered.size());
assertEquals(List.of(), uncovered,
"these keys are in ConfigRef.DEFERRED_KEYS but mutating them alone produces no "
+ "matching entry from changedDeferredKeys — a set entry with no comparison "
+ "behind it, exactly the fleetd #333 F2 shape, confirmed here for "
+ "DEFERRED_KEYS by fleetd #337: " + uncovered);
}
}
@@ -804,7 +804,28 @@ class CompletionResolverTest {
// --- fleetd#201 Unit 1: target-keyed backend-error pattern + typed sink ----------------------
@Test
void aConfiguredBackendErrorPatternClassifiesAMatchAsAFailureAndNotifiesTheSinkOnce() {
void aNormalMemberReportMentioningTheFallbackErrorPatternFailsButDoesNotNotifyTheSink() {
String block = "⏺ I checked the retry path. An API Error: makes it back off.\n❯ ";
FakeHerdr herdr = new FakeHerdr().readText(block);
Rendezvous rendezvous = new Rendezvous();
java.util.List<String> notified = new java.util.ArrayList<>();
BackendErrorSink sink = (target, matchedLine, reason) -> notified.add(target + ": " + matchedLine);
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none(), BackendErrorPatternLookup.legacy(), sink);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
"a scrape mentioning the pattern must still fail the send");
assertTrue(waiter.getNow(null).text().contains("I checked the retry path. An API Error: makes it back off."),
"the failure must keep the whole pane tail");
assertTrue(notified.isEmpty(),
"a normal report mentioning the fallback pattern must not record a credential failure");
}
@Test
void aConfiguredBackendErrorPatternAtTheStartOfALineClassifiesAMatchAndNotifiesTheSinkOnce() {
String block = "⏺ 503 Service Unavailable: upstream credential rejected\n❯ ";
FakeHerdr herdr = new FakeHerdr().readText(block);
Rendezvous rendezvous = new Rendezvous();
@@ -924,7 +945,7 @@ class CompletionResolverTest {
}
@Test
void aConfiguredPatternAlsoClassifiesTheRawScrapeFallbackAndNotifiesTheSink() {
void aConfiguredPatternAtTheStartOfALineAlsoClassifiesTheRawScrapeFallbackAndNotifiesTheSink() {
// No ⏺ marker and leading TUI chrome ⇒ lastAssistantBlock() yields "", so classification must
// fall back to the raw scrape (fleetd#211) — and it must use the configured pattern too.
String block = """
@@ -949,6 +970,40 @@ class CompletionResolverTest {
assertTrue(notified.get(0).contains("503 Service Unavailable"), notified.get(0));
}
/**
* fleetd #339 follow-up: a genuine backend error rendered behind terminal chrome must still
* record the credential outage. #339 added a start-of-line check to stop a member's own prose
* being counted as an outage, and a bare {@code lookingAt} also rejected this — the send failed
* but the sink never fired. #339's invariant 3 named that direction as the worse one: a real
* outage going unrecorded leaves the fleet spawning into a dead credential.
*
* <p>The raw-scrape path is where this matters, because its own comment says to expect leading
* TUI chrome there.
*/
@Test
void aRealErrorBehindTerminalChromeStillNotifiesTheSink() {
String block = """
╭──────────────────────────────────────╮
│ 503 Service Unavailable: upstream credential rejected
""";
FakeHerdr herdr = new FakeHerdr().readText(block);
Rendezvous rendezvous = new Rendezvous();
BackendErrorPatternLookup patterns = target -> Pattern.compile("(?i)503 Service Unavailable");
java.util.List<String> notified = new java.util.ArrayList<>();
BackendErrorSink sink = (target, matchedLine, reason) -> notified.add(target + ": " + matchedLine);
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none(), patterns, sink);
var waiter = rendezvous.open("term_a");
resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
"a real backend error must still fail the send");
assertEquals(1, notified.size(),
"a real error line behind box chrome is still a real outage — it must reach the sink, "
+ "or the fleet keeps spawning into a dead credential");
}
// --- fleetd#201 Unit 1: classification inside the fleetd#164 MIN_TURN_NANOS floor -------------
@Test
@@ -48,7 +48,7 @@ class InjectorTest {
@Test
void deliversWhenIdle() {
CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T)).completion();
assertFalse(f.isDone(), "not delivered until an injectable status arrives");
injector.onStatus(T, AgentStatus.IDLE);
assertTrue(f.isDone());
@@ -170,6 +170,32 @@ class InjectorTest {
assertEquals(List.of("a", "b", "c"), sent());
}
@Test
void cancellingTheMiddleDeliveryKeepsTheFollowingDeliveryReachable() {
Injector.Delivery first = injector.enqueue(T, "same text", TestTurnTokens.inert(T));
Injector.Delivery cancelled = injector.enqueue(T, "same text", TestTurnTokens.inert(T));
injector.enqueue(T, "after cancelled", TestTurnTokens.inert(T));
assertEquals(Injector.Cancellation.CANCELLED, injector.cancel(cancelled));
injector.onStatus(T, AgentStatus.IDLE);
injector.onStatus(T, AgentStatus.WORKING);
injector.onStatus(T, AgentStatus.IDLE);
assertEquals(List.of("same text", "after cancelled"), sent(),
"cancellation must match the exact Delivery and preserve the remaining FIFO queue");
assertTrue(first.completion().isDone());
}
@Test
void cancellationReportsDeliveredWhenPickupWonTheRace() {
Injector.Delivery delivery = injector.enqueue(T, "already sent", TestTurnTokens.inert(T));
injector.onStatus(T, AgentStatus.IDLE);
assertEquals(Injector.Cancellation.DELIVERED, injector.cancel(delivery),
"a cancellation after pickup must not claim that the text stayed queued");
assertEquals(List.of("already sent"), sent());
}
@Test
void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
assertTrue(injector.activeTargets().isEmpty());
@@ -378,7 +404,7 @@ class InjectorTest {
void sendFailureDropsMessageAndFailsItsFuture() {
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
Injector inj = new Injector(new AgentControl(failing));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T)).completion();
inj.onStatus(T, AgentStatus.IDLE);
assertTrue(f.isCompletedExceptionally());
@@ -387,7 +413,7 @@ class InjectorTest {
@Test
void dropFailsPendingWaiters() {
CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T)).completion();
injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
}
@@ -396,8 +422,8 @@ class InjectorTest {
void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), cap);
CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));
CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T)).completion();
CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T)).completion();
inj.onStatus(T, AgentStatus.IDLE); // deliver the first message
inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
@@ -449,7 +475,7 @@ class InjectorTest {
Captor cap = new Captor();
List<String> forgotten = new ArrayList<>();
Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T)).completion();
for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
@@ -563,7 +589,7 @@ class InjectorTest {
StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
poller.start();
try {
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T)).completion();
delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
} finally {
poller.stop();
@@ -582,7 +608,7 @@ class InjectorTest {
void deliveredFutureCarriesSendFailure() {
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
Injector inj = new Injector(new AgentControl(failing));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T)).completion();
inj.onStatus(T, AgentStatus.IDLE);
ExecutionException ex = assertThrows(ExecutionException.class, f::get);
assertInstanceOf(HerdrException.class, ex.getCause());
@@ -41,7 +41,7 @@ class StatusPollerRoutingTest {
poller.start();
try {
CompletableFuture<Void> delivered =
injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET));
injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET)).completion();
// Must resolve quickly: refining against the WRONG daemon (member) never classifies
// out of UNKNOWN, so this would time out under the bug.
delivered.get(2, TimeUnit.SECONDS);
@@ -65,7 +65,7 @@ class StatusPollerRoutingTest {
poller.start();
try {
CompletableFuture<Void> delivered =
injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET));
injector.enqueue(LEAD_TARGET, "via-poller", TestTurnTokens.inert(LEAD_TARGET)).completion();
assertThrows(TimeoutException.class, () -> delivered.get(500, TimeUnit.MILLISECONDS),
"a lead target must never be refined from the member daemon's pane content");
} finally {
@@ -615,6 +615,67 @@ class CompositePeerLauncherTest {
assertEquals(1, adapter.spawnCount("b"));
}
/**
* fleetd #315: {@code CompositePeerLauncher.spawn} rebuilds the {@link PlacementContext} after
* every failed attempt "so the policy excludes this profile" (see the comment at the retry call
* site) — but {@code FixedPlacementPolicy} never read {@code ctx.unreachable()}, so under the
* default {@code fixed} placement every retry re-picked the same dead default and a second,
* healthy, configured profile was never tried. This is the same scenario as
* {@link #failoverRetriesNextCandidateWhenProfileIsUnreachable}, but pinned to {@code fixed()}
* instead of {@code weighted()} — the three existing failover tests all use {@code weighted()},
* which is exactly why nobody caught this: the retry loop's contract has no coverage under its
* own default policy.
*/
@Test
void failoverRetriesNextCandidateUnderFixedPlacementWhenProfileIsUnreachable() {
FakeHerdr herdr = new FakeHerdr();
Map<String, FleetConfig.Profile> profiles = ordered(
"a", stubWorker("a"),
"b", stubWorker("b"));
StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "a", Set.of("a"));
CompositePeerLauncher composite = new CompositePeerLauncher(
List.of(adapter), "a", profiles, PlacementPolicies.fixed(), _ -> 0);
PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
assertEquals("b", h.profile(),
"fixed placement must fail over from the unreachable default a to the healthy b");
assertEquals(1, adapter.spawnCount("a"), "a was tried once and failed");
assertEquals(1, adapter.spawnCount("b"), "b was tried once and succeeded");
}
/**
* fleetd #315: the same fix — {@code FixedPlacementPolicy} consulting {@code ctx.unreachable()}
* — also covers the wiring-bug branch in {@code CompositePeerLauncher.spawn}: a profile that
* placement is allowed to choose (it is in the configured candidate list) but that no delegate
* declares ({@code byProfile.get(chosen.profile()) == null}). That branch adds the profile to
* {@code unreachable} and {@code continue}s without ever calling a launcher, so before this fix
* {@code fixed} handed back the same adapterless profile on every remaining attempt too.
*/
@Test
void failoverSkipsAConfiguredProfileNoAdapterDeclaresUnderFixedPlacement() {
FakeHerdr herdr = new FakeHerdr();
// Placement's candidate list has three profiles, in this order (LinkedHashMap preserves it,
// and the fixed default resolves to the first — see the `ordered` helper's own javadoc).
Map<String, FleetConfig.Profile> profiles = new LinkedHashMap<>();
profiles.put("c", stubWorker("c"));
profiles.put("a", stubWorker("a"));
profiles.put("b", stubWorker("b"));
// The adapter only declares a and b — c is a configured profile with no owning adapter,
// the "wiring bug" the comment in CompositePeerLauncher.spawn calls out.
Map<String, FleetConfig.Profile> adapterProfiles = new LinkedHashMap<>();
adapterProfiles.put("a", stubWorker("a"));
adapterProfiles.put("b", stubWorker("b"));
StubLauncher adapter = new StubLauncher("claude", herdr, adapterProfiles, "a", Set.of());
CompositePeerLauncher composite = new CompositePeerLauncher(
List.of(adapter), "a", profiles, PlacementPolicies.fixed(), _ -> 0);
PeerHandle h = composite.spawn(new SpawnRequest(null, null, null));
assertEquals("a", h.profile(),
"c has no adapter, so fixed placement must skip it and land on the next candidate, a");
assertEquals(0, adapter.spawnCount("c"), "c is never spawned — no adapter owns it");
assertEquals(1, adapter.spawnCount("a"));
}
@Test
void explicitSpawnAtMaxLoadThrowsPlacementExceptionNamingProfileLiveAndCap() {
FakeHerdr herdr = new FakeHerdr();
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -370,6 +371,161 @@ class HerdrPeerLauncherAllowListWiringTest {
"expected the pre-existing 'scrub blanks them' INFO unchanged, got: " + messages);
}
/**
* fleetd #341: {@code unprotectedGapLogged} guarded TWO WARN branches that name DIFFERENT env
* var names — the allow-list branch ({@code keptByDerivedList}, below) and the deny-by-default
* / non-zsh-fallback branch ({@link HerdrPeerLauncher#warnGapUnprotected}). {@code
* memberCredentials} is a live, re-read-per-spawn supplier, so the policy can change between
* two spawns on the same launcher instance — a config reload needs no restart. Spawn 1 runs
* under {@code deny-by-default} with a gap of {@code SPAWN_ONE_UNCOVERED_TOKEN}, which trips
* the (before this fix) SHARED one-shot flag. The policy is then reloaded to {@code
* allow-list}; spawn 2's gap is {@code FLEETD_WORKER_TOKEN} instead — the test profile's own
* {@code tokenEnv}, which the derived allow-list keeps even though it is on neither {@code
* known:} nor {@code allow:}, so it is genuinely unprotected and deserves its own WARN. Before
* this fix that WARN never fires, because the shared flag was already {@code true} — the
* operator is never told {@code FLEETD_WORKER_TOKEN} reaches every member pane unblocked. Real
* path: two real {@link HerdrPeerLauncher#spawn} calls on ONE launcher instance, with mutable
* {@code memberCredentials}/host-env suppliers standing in for a live config reload between
* spawns.
*/
@Test
void aDifferentUnprotectedGapOnALaterSpawnIsNotSuppressedByAnEarlierSpawnsWarn() {
FakeHerdr herdr = new FakeHerdr();
AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
new FleetConfig.MemberCredentials(null, List.of(), List.of(), null)); // deny-by-default
AtomicReference<Set<String>> hostEnvState =
new AtomicReference<>(Set.of("SPAWN_ONE_UNCOVERED_TOKEN"));
WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
Level original = logger.getLevel();
logger.setLevel(Level.WARN);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
try {
// Spawn 1: deny-by-default, gap = {SPAWN_ONE_UNCOVERED_TOKEN} — the effectiveAllowed ==
// null branch, via warnGapUnprotected.
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
// Live policy reload to allow-list, with a DIFFERENT gap name.
credsState.set(new FleetConfig.MemberCredentials(
FleetConfig.MemberCredentials.POLICY_ALLOW_LIST, List.of(), List.of(), null));
hostEnvState.set(Set.of("FLEETD_WORKER_TOKEN"));
// Spawn 2: allow-list, gap = {FLEETD_WORKER_TOKEN} — kept by the derived allow-list
// (the profile's own tokenEnv), so it is the effectiveAllowed != null / keptByDerivedList
// branch, at the SAME log line HerdrPeerLauncher:1819 guards with the shared flag.
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
} finally {
logger.detachAppender(appender);
logger.setLevel(original);
}
List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
assertTrue(messages.stream().anyMatch(
m -> m.contains("UNBLOCKED") && m.contains("SPAWN_ONE_UNCOVERED_TOKEN")),
"spawn 1's deny-by-default gap must still warn — got: " + messages);
assertTrue(messages.stream().anyMatch(
m -> m.contains("UNBLOCKED") && m.contains("FLEETD_WORKER_TOKEN")),
"spawn 2's gap names a DIFFERENT env var than spawn 1 (FLEETD_WORKER_TOKEN, not "
+ "SPAWN_ONE_UNCOVERED_TOKEN) — it must still be warned about even though a "
+ "flag already fired once for spawn 1's unrelated name. Before fleetd #341's "
+ "fix this WARN never fires because unprotectedGapLogged was already true. "
+ "Got: " + messages);
}
/**
* fleetd #341 follow-up: the OTHER half of the guard's contract. The set exists to report every
* distinct name, but it must still report each one only ONCE — the noise control is the reason
* a guard is here at all, and the ticket named it as invariant 1. Two spawns, same policy, same
* gap name: exactly one WARN mentioning it.
*
* <p>Measured before this test existed: replacing {@code .filter(unprotectedGapNamesWarned::add)}
* with a filter that adds and always returns {@code true} — so every name is logged on every
* spawn — left all 1358 tests green. The fix was correct and nothing held it there. That is the
* "a test on the seam does not prove the caller" shape: the {@code Set} behaves, and nothing
* proved this class used it as a guard rather than as a record.
*/
@Test
void theSameUnprotectedNameIsWarnedAboutOnlyOnceAcrossSpawns() {
FakeHerdr herdr = new FakeHerdr();
AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
new FleetConfig.MemberCredentials(null, List.of(), List.of(), null)); // deny-by-default
AtomicReference<Set<String>> hostEnvState =
new AtomicReference<>(Set.of("REPEATED_UNCOVERED_TOKEN"));
WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
Level original = logger.getLevel();
logger.setLevel(Level.WARN);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
try {
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
// Same policy, same gap, second spawn. Nothing new to tell the operator.
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
} finally {
logger.detachAppender(appender);
logger.setLevel(original);
}
List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
long mentioning = messages.stream()
.filter(m -> m.contains("REPEATED_UNCOVERED_TOKEN"))
.count();
assertEquals(1, mentioning,
"one unchanged unprotected name across two spawns must produce exactly one WARN — "
+ "the set is a guard, not just a record. Got: " + messages);
}
/**
* fleetd #341 follow-up: the reverse policy order. The original defect was found going
* deny-by-default then allow-list, and a guard that is fixed in one direction is not
* necessarily fixed in the other — "ask which states still OPEN the gate". Here spawn 1 runs
* under {@code allow-list} (the {@code keptByDerivedList} branch) and spawn 2 under
* {@code deny-by-default} ({@link HerdrPeerLauncher#warnGapUnprotected}), with a different name
* each time. Both must be reported.
*/
@Test
void anAllowListWarnDoesNotSuppressALaterDenyByDefaultWarnForADifferentName() {
FakeHerdr herdr = new FakeHerdr();
AtomicReference<FleetConfig.MemberCredentials> credsState = new AtomicReference<>(
new FleetConfig.MemberCredentials(
FleetConfig.MemberCredentials.POLICY_ALLOW_LIST, List.of(), List.of(), null));
AtomicReference<Set<String>> hostEnvState =
new AtomicReference<>(Set.of("FLEETD_WORKER_TOKEN"));
WiringLauncher launcher = new WiringLauncher(herdr, credsState::get, "/bin/zsh", hostEnvState::get);
Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
Level original = logger.getLevel();
logger.setLevel(Level.WARN);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
try {
// Spawn 1: allow-list, gap kept by the derived list — the keptByDerivedList WARN.
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
// Live reload the OTHER way: back to deny-by-default, with a different name.
credsState.set(new FleetConfig.MemberCredentials(null, List.of(), List.of(), null));
hostEnvState.set(Set.of("LATER_UNCOVERED_TOKEN"));
launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
} finally {
logger.detachAppender(appender);
logger.setLevel(original);
}
List<String> messages = appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
assertTrue(messages.stream().anyMatch(
m -> m.contains("UNBLOCKED") && m.contains("FLEETD_WORKER_TOKEN")),
"spawn 1's allow-list gap must warn — got: " + messages);
assertTrue(messages.stream().anyMatch(
m -> m.contains("UNBLOCKED") && m.contains("LATER_UNCOVERED_TOKEN")),
"spawn 2's deny-by-default gap names a different variable and must still be warned "
+ "about, even though an allow-list WARN already fired. Got: " + messages);
}
/**
* fleetd #185 stage 2: with {@code memberHerdrSocket:} configured, member panes run under a
* different OS user — {@link HerdrPeerLauncher#hostEnvNames} describes fleetd's own process, not
@@ -0,0 +1,264 @@
package dev.ltms.fleet.msg;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import com.rabbitmq.client.Envelope;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* CB-318: a delivery landing on the consumer work-pool thread <em>after</em>
* {@link AmqpReplyInbox#release} has already swapped the target's {@code held} entry for its
* tombstone, but <em>before</em> {@code release()} itself returns, must be nacked-with-requeue —
* never silently retained in a fresh map {@code release()} has already stopped looking at.
*
* <p><strong>This forces the actual interleaving, not a sequence of calls.</strong> {@code release()}
* runs on its own thread and is made to block <em>inside</em> its nack loop, via a fake
* {@link Channel} whose {@code basicNack} blocks on its first invocation. That block is only
* reachable after {@code release()}'s {@code held.compute(...)} has already swapped in the
* {@code RELEASED} tombstone (the compute call happens strictly before the loop that calls
* {@code basicNack}), so observing it is direct, ordering-guaranteed proof that the tombstone is in
* place and {@code release()} has not yet returned — still holding {@code channelLock} — when a
* second thread fires {@code own()}'s captured {@link DeliverCallback} for a brand-new message on the
* same target. No mocking library is on the classpath, so the fake broker is a {@link Proxy}, the
* same pattern {@code AmqpReplyInboxRecoveryRaceTest} already uses.
*
* <p><strong>What this does and does not prove.</strong> It proves that a delivery whose
* {@code computeIfAbsent} call is ordered strictly after {@code release()}'s tombstone swap — while
* {@code release()} is still running — is nacked-with-requeue rather than silently parked forever.
* It does not drive a real broker: {@code basicNack} here is a recorded call on a fake channel, not a
* verified requeue-and-redeliver. That half of the contract (a nacked-with-requeue delivery really
* does come back to a later owner) is already covered against a real broker by
* {@code AmqpReplyInboxContractTest.releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery}, which
* this test does not duplicate.
*/
class AmqpReplyInboxReleaseRaceTest {
@Test
@Timeout(15)
void deliveryArrivingWhileReleaseIsStillRunningIsNackedNotStranded() throws Exception {
String target = "worker-release-race";
List<long[]> nacks = new CopyOnWriteArrayList<>(); // {deliveryTag, requeue(1/0)}
List<Long> acks = new CopyOnWriteArrayList<>();
AtomicReference<DeliverCallback> deliverCallback = new AtomicReference<>();
CountDownLatch nackStarted = new CountDownLatch(1);
CountDownLatch releaseMayFinishNack = new CountDownLatch(1);
AtomicInteger nackCallCount = new AtomicInteger();
Channel consumeChannel = fakeConsumeChannel(deliverCallback, nacks, acks, nackCallCount,
nackStarted, releaseMayFinishNack);
Channel publishChannel = fakeInertChannel();
Connection connection = fakeConnection(consumeChannel, publishChannel);
AmqpReplyInbox inbox = new AmqpReplyInbox(connection, AmqpReplyInbox.DEFAULT_PREFETCH);
inbox.own(target);
assertTrue(deliverCallback.get() != null, "own() must have registered a DeliverCallback");
// Seed one already-held delivery (m0) so release()'s nack loop has something to iterate, and
// therefore somewhere to block, before it can return.
deliverCallback.get().handle("ctag", delivery(1L, "m0", "first"));
AtomicReference<Throwable> releaseError = new AtomicReference<>();
Thread releaseThread = new Thread(() -> {
try {
inbox.release(target);
} catch (Throwable t) {
releaseError.set(t);
}
}, "release-under-test");
releaseThread.start();
// This latch only fires from inside the fake channel's basicNack — i.e. from inside
// release()'s nack loop, which release()'s code only reaches AFTER held.compute(...) has
// already swapped in RELEASED. Waiting for it is direct proof the swap has happened and
// release() has not yet returned (it is stuck mid-loop, still holding channelLock).
assertTrue(nackStarted.await(10, TimeUnit.SECONDS),
"release() never reached its nack loop — it may not have started");
// The exact interleaving CB-318 describes: a delivery for a NEW message on the same target
// lands on the "consumer work-pool thread" (this second thread) while release() is still
// running. With the pre-fix code (a bare held.remove(target)) this created a brand-new map
// under computeIfAbsent that release() — already past its remove — never looks at again.
AtomicReference<Throwable> deliveryError = new AtomicReference<>();
Thread deliveryThread = new Thread(() -> {
try {
deliverCallback.get().handle("ctag", delivery(2L, "m1", "second"));
} catch (Throwable t) {
deliveryError.set(t);
}
}, "concurrent-delivery");
deliveryThread.start();
// Head start for the delivery thread to reach (and, on the fixed code, block on)
// channelLock — release() still holds it at this point, so a correct fix cannot have
// resolved m1's nack yet. Purely in-memory work (computeIfAbsent, a reference compare)
// separates deliveryThread.start() from that block point, so 300ms is a large margin, not a
// tight timing assumption.
Thread.sleep(300);
assertEquals(1, nackCallCount.get(),
"the concurrent delivery must not resolve its nack before release() gives up "
+ "channelLock — if this is 2 already, the interleaving below is not being "
+ "tested, only a sequential call");
releaseMayFinishNack.countDown(); // let release() finish nacking m0 and return
assertTrue(releaseThread.join(Duration.ofSeconds(10)), "release() did not finish");
assertTrue(deliveryThread.join(Duration.ofSeconds(10)), "the concurrent delivery did not finish");
assertNull(releaseError.get(), "release() threw: " + releaseError.get());
assertNull(deliveryError.get(), "the concurrent delivery threw: " + deliveryError.get());
assertEquals(2, nacks.size(),
"both the pre-held m0 and the concurrently-arriving m1 must be nacked, got: "
+ nacks.stream().map(n -> "[tag=" + n[0] + " requeue=" + n[1] + "]").toList());
assertTrue(nacks.stream().allMatch(n -> n[1] == 1L),
"invariant 1 (never drop): every nack must set requeue=true");
assertTrue(nacks.stream().anyMatch(n -> n[0] == 1L), "m0's delivery tag must be nacked");
assertTrue(nacks.stream().anyMatch(n -> n[0] == 2L),
"m1 — delivered while release() was still running, after the tombstone swap — must be "
+ "nacked, not silently retained in a map release() will never look at again");
assertTrue(acks.isEmpty(), "invariant 1 (never drop): a held reply must never be basicAck'd");
}
private static Delivery delivery(long tag, String msgId, String body) {
Envelope envelope = new Envelope(tag, false, "", "irrelevant");
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();
return new Delivery(envelope, props, body.getBytes(StandardCharsets.UTF_8));
}
/** A {@link Proxy}-backed consume {@link Channel}: blocks the FIRST {@code basicNack} call on
* {@code releaseMayFinishNack}, after signalling {@code nackStarted} — everything else records
* the call and returns a harmless default, matching the style already used by
* {@code AmqpReplyInboxRecoveryRaceTest}. */
private static Channel fakeConsumeChannel(AtomicReference<DeliverCallback> deliverCallback,
List<long[]> nacks, List<Long> acks,
AtomicInteger nackCallCount,
CountDownLatch nackStarted,
CountDownLatch releaseMayFinishNack) {
InvocationHandler handler = (proxy, method, args) -> {
String name = method.getName();
if (name.equals("basicConsume")) {
deliverCallback.set((DeliverCallback) args[2]);
return "ctag";
}
if (name.equals("basicNack")) {
long tag = (long) args[0];
boolean requeue = (boolean) args[2];
if (nackCallCount.incrementAndGet() == 1) {
nackStarted.countDown();
if (!releaseMayFinishNack.await(10, TimeUnit.SECONDS)) {
throw new IllegalStateException("test never released the nack latch");
}
}
nacks.add(new long[] {tag, requeue ? 1L : 0L});
return null;
}
if (name.equals("basicAck")) {
acks.add((long) args[0]);
return null;
}
if (name.equals("equals")) {
return proxy == args[0];
}
if (name.equals("hashCode")) {
return System.identityHashCode(proxy);
}
if (name.equals("toString")) {
return "FakeConsumeChannel";
}
return defaultValue(method.getReturnType());
};
return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
new Class<?>[] {Channel.class}, handler);
}
/** A {@link Proxy}-backed {@link Channel} that answers every call with a harmless default — used
* as the publish channel, which this test never actually publishes on. */
private static Channel fakeInertChannel() {
InvocationHandler handler = (proxy, method, args) -> {
String name = method.getName();
if (name.equals("equals")) {
return proxy == args[0];
}
if (name.equals("hashCode")) {
return System.identityHashCode(proxy);
}
if (name.equals("toString")) {
return "FakeInertChannel";
}
return defaultValue(method.getReturnType());
};
return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
new Class<?>[] {Channel.class}, handler);
}
/** A {@link Proxy}-backed {@link Connection} handing out {@code first} then {@code second} from
* successive {@code createChannel()} calls, matching {@link AmqpReplyInbox}'s constructor. */
private static Connection fakeConnection(Channel first, Channel second) {
AtomicInteger calls = new AtomicInteger();
InvocationHandler handler = (proxy, method, args) -> {
String name = method.getName();
if (name.equals("createChannel") && (args == null || args.length == 0)) {
return calls.getAndIncrement() == 0 ? first : second;
}
if (name.equals("equals")) {
return proxy == args[0];
}
if (name.equals("hashCode")) {
return System.identityHashCode(proxy);
}
if (name.equals("toString")) {
return "FakeConnection";
}
return defaultValue(method.getReturnType());
};
return (Connection) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
new Class<?>[] {Connection.class}, handler);
}
private static Object defaultValue(Class<?> type) {
if (!type.isPrimitive() || type == void.class) {
return null;
}
if (type == boolean.class) {
return Boolean.FALSE;
}
if (type == long.class) {
return 0L;
}
if (type == short.class) {
return (short) 0;
}
if (type == byte.class) {
return (byte) 0;
}
if (type == char.class) {
return (char) 0;
}
if (type == double.class) {
return 0.0d;
}
if (type == float.class) {
return 0.0f;
}
return 0;
}
}
@@ -1,5 +1,9 @@
package dev.ltms.fleet.msg;
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.herdr.AgentControl;
import dev.ltms.fleet.herdr.AgentStatus;
import dev.ltms.fleet.herdr.FakeHerdr;
@@ -11,6 +15,7 @@ import dev.ltms.fleet.mcp.PrimaryRegistry;
import dev.ltms.fleet.inject.Injector;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -253,7 +258,7 @@ class MessageServiceTest {
injector.onStatus(T, AgentStatus.IDLE); // first delivery
injector.onStatus(T, AgentStatus.WORKING); // first turn in flight
CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T)).completion();
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
@@ -405,6 +410,29 @@ class MessageServiceTest {
"a delivered send whose worker never replies times out as still working");
}
/**
* fleetd #345. This forces the injector to pick up the exact pending delivery after {@code send}
* first observes its completion as incomplete, but before {@code cancel} takes the target monitor.
* The timeout must use {@link Injector.Cancellation#DELIVERED} from {@code cancel} and report
* {@link MessageService.Outcome#TIMED_OUT_WORKING}, because the text landed.
*
* <p>What this does not prove: that this precise interleaving happens by itself under production
* timing. The test forces it through a test-only hook; it proves the timeout caller handles the
* injector result when the interleaving occurs.
*/
@Test
void sendTimeoutUsesCancellationDeliveredWhenPickupWinsTheRace() {
messages.setTimeoutCancellationRaceHookForTest(() -> injector.onStatus(T, AgentStatus.IDLE));
try {
MessageService.Reply reply = messages.send(T, "race delivery", 50);
assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, reply.outcome(),
"cancel reporting DELIVERED means the worker received the timed-out message");
} finally {
messages.setTimeoutCancellationRaceHookForTest(null);
}
}
@Test
void answerTimesOutWhenTheResumedWorkerNeverReplies() throws Exception {
CompletableFuture<MessageService.Reply> send = sendAsync();
@@ -940,6 +968,54 @@ class MessageServiceTest {
assertEquals("PR opened: https://example/pulls/42", view.reply());
}
/**
* fleetd #324: {@code answer()} holds {@code sessionLocks} for the target and, once the worker's
* real terminal reply arrives, calls {@code finishAsyncTask}, which used to read the volatile
* {@code task.turnId} twice — once to check it is non-null, once as the key for
* {@code asyncTasksByTurn.remove}. {@code ask()}'s own timeout path mutates the same field with no
* lock at all. This test does not wait for a real race to land in that narrow window between the
* two reads — instead it drives the exact sequence the ticket describes (worker asks, primary
* answers, worker's real reply arrives) and, via a package-private test hook wired to fire at
* precisely that point, runs the identical production cleanup {@code ask()}'s timeout catch block
* runs ({@code clearAsyncQuestion(turnId, true)}) so the field goes {@code null} between the two
* reads deterministically rather than by chance.
*
* <p>What this proves: given that exact interleaving, {@code answer()} must not throw and the
* ticket must still resolve to the worker's real reply. What it does not prove: that the
* interleaving itself is reachable in production — that is established by reading the code (see
* the ticket), not by this test, since forcing it via a hook is not the same as two independent
* threads racing on their own schedules.
*/
@Test
void finishAsyncTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
String turnId = asking.turnId();
// Fire ask()'s own unlocked timeout cleanup at the moment finishAsyncTask has already checked
// task.turnId is non-null but has not yet used it — the exact torn-read window fleetd #324
// describes.
messages.setFinishAsyncTaskRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
CompletableFuture<MessageService.Reply> answer =
CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
awaitWaiting(); // answer() opened its own forward waiter for the resumed worker turn
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
"the lead's own answer() call must not throw because ask()'s timeout cleanup raced it");
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("PR opened: https://example/pulls/42", done.reply(),
"the ticket must still resolve to the worker's real reply despite the forced race");
}
@Test
void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
@@ -993,6 +1069,105 @@ class MessageServiceTest {
"the reply completed its own ticket directly and never touched the inbox");
}
/**
* fleetd #329 (F1). {@code answer()} completes the async ticket by looking {@code turnId} up in
* {@code asyncTasksByTurn} a SECOND time (the first is at :991, purely to re-register the
* {@code asyncTasksByWaiter} entry for #282's chained-ask case). That second lookup races
* {@code ask()}'s own unlocked timeout cleanup ({@code clearAsyncQuestion(turnId, true)}, run from
* {@code markAskTimedOut} + forgetting): {@code ask()}'s {@code ticket.answer().get(timeoutMillis)}
* can time out at essentially the same instant {@code answer()}'s {@code rendezvous.answerAsk}
* call above already succeeded and unblocked the worker. fleetd #324's single-read fix does not
* help here — that fixed a torn read of one already-held {@link MessageService} internal
* {@code Task}; this is a second, independent map lookup by a caller that no longer holds the
* {@code Task} it already found once.
*
* <p>This test does not wait for that real race to land on its own schedule — it drives the exact
* sequence the ticket describes (worker asks, primary answers, worker's real reply arrives) and
* fires the identical production cleanup {@code ask()}'s timeout path runs
* ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd #324
* already merged) at the point between "the worker's own {@code ask()} call has unblocked" and
* "the worker's real {@code fleet_reply} arrives" — the exact window fleetd #329 names.
*
* <p>What this proves: given that exact interleaving, the async ticket must still resolve to the
* worker's real reply, not stay stuck {@link MessageService.Phase#PENDING} forever. What it does
* not prove: that the interleaving itself is reachable in production on its own timing — that is
* established by reading the code (see the ticket's "path in"), not by this test, for the same
* reason fleetd #324's own race test says so.
*/
@Test
void aReplyRacingAsksTimeoutCleanupStillCompletesTheAsyncTicket() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
String turnId = asking.turnId();
CompletableFuture<MessageService.Reply> answer =
CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer(),
"the worker's own ask() call must have already unblocked with the primary's answer "
+ "before we force the race below");
// ask()'s own unlocked timeout cleanup can forget this exact turnId at essentially the same
// instant answer() has already unblocked the worker and is now waiting on the resumed turn's
// real reply — reproduce that interleaving directly instead of trying to win a real race.
messages.forgetTurnForTest(turnId);
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
"the primary's own answer() call must still see the worker's real reply");
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 "
+ "just because ask()'s timeout cleanup forgot this turnId first");
}
/**
* fleetd #329 (F3). {@link MessageService#reply} reads the same orphan {@code Task}'s {@code
* turnId} twice on this path (once to check it is non-null, once as the {@code
* asyncTasksByTurn.remove} key) — the exact double-read shape fleetd #324 fixed in {@code
* finishAsyncTask}. This test drives the same real sequence as {@code
* aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket} (worker asks, primary answers, the
* primary's own bounded wait for the resumed turn expires) to reach a task with {@code turnId}
* genuinely stamped and no live rendezvous waiter open — the state {@code askAnsweredAsyncTasks}
* matches here — then fires the identical production cleanup {@code ask()}'s own timeout path
* runs ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd
* #324 merged) at the point between the check and the removal use, via a dedicated test hook
* mirroring {@code finishAsyncTaskRaceHook}.
*/
@Test
void replyToAnOrphanedTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
String turnId = asking.turnId();
MessageService.Reply answerReply = messages.answer(turnId, "config.yaml", 150);
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
"the primary's own bounded wait must give up first, leaving turnId stamped with no "
+ "live waiter — the state reply()'s F3 code path matches");
messages.setReplyOrphanTurnIdRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
try {
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(),
"the ticket must still resolve to the worker's real reply despite the forced race");
} finally {
messages.setReplyOrphanTurnIdRaceHookForTest(null);
}
}
/**
* 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
@@ -1103,6 +1278,63 @@ class MessageServiceTest {
assertEquals("done", done.reply());
}
// --- fleetd #329 (F2): an exception after the ticket's future completes must reach a log -----
//
// sendAsync's own executor task ends with catch (Throwable t) { task.future.completeExceptionally(t); }
// — but finishAsyncTask completes that same future on its first line, so anything that throws
// afterward hits an already-completed future: completeExceptionally returns false and does
// nothing, and (before this fix) nothing logged it either. Measured on the ticket: temporarily
// reintroducing the fleetd #324 NPE reproduced 19 real exceptions on the ordinary path with
// 74/74 tests staying green and zero log lines. There is no reachable production call site where
// finishAsyncTask throws after completing the future (fleetd #324 already closed the one that
// used to), so this test drives the shape directly via a dedicated test-only hook
// (afterFinishAsyncTaskCompleteHookForTest) rather than trying to engineer a real exception into
// that narrow window — the same technique fleetd #324's own race test and this ticket's F1/F3
// tests use for their own races.
private static ListAppender<ILoggingEvent> attachMessageServiceLog() {
Logger logger = (Logger) LoggerFactory.getLogger(MessageService.class);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
return appender;
}
private static void detachMessageServiceLog(ListAppender<ILoggingEvent> appender) {
((Logger) LoggerFactory.getLogger(MessageService.class)).detachAppender(appender);
}
@Test
void anExceptionAfterTheTicketFutureCompletesStillReachesTheLog() throws Exception {
ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
try {
messages.setAfterFinishAsyncTaskCompleteHookForTest(() -> {
throw new RuntimeException("PROBE-329-F2");
});
String ticket = messages.sendAsync(T, "do the task");
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
// finishAsyncTask's own first line already completed the future with the real result
// BEFORE the hook threw — F2 is a visibility gap, not a correctness gap for the ticket
// itself, so the ticket's own outcome must be unaffected by the swallowed exception.
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("done", done.reply());
assertTrue(appender.list.stream().anyMatch(e ->
e.getLevel() == Level.ERROR
&& e.getFormattedMessage().contains(ticket)
&& e.getThrowableProxy() != null
&& "PROBE-329-F2".equals(e.getThrowableProxy().getMessage())),
"an exception thrown after the ticket's future already completed must still "
+ "reach the log, not vanish silently");
} finally {
messages.setAfterFinishAsyncTaskCompleteHookForTest(null);
detachMessageServiceLog(appender);
}
}
// --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
@Test
@@ -1559,7 +1791,10 @@ class MessageServiceTest {
assertEquals(MessageService.Outcome.TIMED_OUT_QUEUED, r.outcome());
assertTrue(messages.hasQueuedDelivery(T),
"a TIMED_OUT_QUEUED send leaves the message still queued in the injector");
"a TIMED_OUT_QUEUED send still records the undelivered delivery for fleet health");
injector.onStatus(T, AgentStatus.IDLE);
assertTrue(herdr.calls.stream().noneMatch(c -> c.method().equals("agent.prompt")),
"a TIMED_OUT_QUEUED send must be cancelled, not delivered when the worker later goes idle");
}
@Test
@@ -86,17 +86,50 @@ class SessionManagerTest {
private volatile RuntimeException hasUncommittedFailure;
private volatile RuntimeException snapshotFailure;
private final java.util.concurrent.atomic.AtomicLong snapshotSeq = new java.util.concurrent.atomic.AtomicLong();
/** fleetd #316: successive {@code hasUncommitted} answers, one per call, last one sticky
* once exhausted — models a worktree whose state changes between reads. Empty (the
* default) falls back to the plain {@link #dirty} flag, so every existing test using this
* fake keeps returning one fixed answer. */
private final List<Boolean> dirtySequence = new java.util.concurrent.CopyOnWriteArrayList<>();
private int failHasUncommittedOnCall = -1;
private RuntimeException hasUncommittedCallFailure;
private final java.util.concurrent.atomic.AtomicInteger hasUncommittedCalls =
new java.util.concurrent.atomic.AtomicInteger();
RecordingWorktrees dirty(boolean dirty) {
this.dirty = dirty;
return this;
}
/** fleetd #316: return {@code answers[0]} on the first {@code hasUncommitted} call,
* {@code answers[1]} on the second, and so on; the last element repeats after that. */
RecordingWorktrees dirtySequence(boolean... answers) {
for (boolean a : answers) {
dirtySequence.add(a);
}
return this;
}
int hasUncommittedCallCount() {
return hasUncommittedCalls.get();
}
RecordingWorktrees failHasUncommittedWith(RuntimeException e) {
this.hasUncommittedFailure = e;
return this;
}
/**
* Throw from {@code hasUncommitted} on one specific call only, counting from 0. The
* whole-double {@link #failHasUncommittedWith} cannot express fleetd #316's fail-safe
* case, which needs the pre-stop read to succeed and only the late read to fail.
*/
RecordingWorktrees failHasUncommittedOnCall(int call, RuntimeException e) {
this.failHasUncommittedOnCall = call;
this.hasUncommittedCallFailure = e;
return this;
}
RecordingWorktrees failRemoveFor(String worktreePath) {
failRemoveFor.add(worktreePath);
return this;
@@ -127,9 +160,16 @@ class SessionManagerTest {
@Override
public boolean hasUncommitted(String worktreePath) {
int call = hasUncommittedCalls.getAndIncrement();
if (hasUncommittedFailure != null) {
throw hasUncommittedFailure;
}
if (call == failHasUncommittedOnCall) {
throw hasUncommittedCallFailure;
}
if (!dirtySequence.isEmpty()) {
return dirtySequence.get(Math.min(call, dirtySequence.size() - 1));
}
return dirty;
}
@@ -1207,6 +1247,104 @@ class SessionManagerTest {
+ "dirty check threw");
}
// --- fleetd #316: the dirty check must be re-taken after the worker is stopped, not trusted
// stale from before it ------------------------------------------------------------------------
@Test
void releaseDoesNotRemoveAWorktreeThatBecameDirtyBetweenTheFirstCheckAndRemoval() {
// Models the exact race #316 reports: hasUncommitted answers clean while the worker is
// still running (call 1), the worker then writes new work, and by the time release is
// about to force-remove the worktree a second read (call 2) would see it as dirty. Without
// the fix this test fails: release() never re-reads and force-removes the worktree anyway.
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees().dirtySequence(false, true);
SessionManager sessions = sessionManager(herdr, worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-316a", null));
sessions.release(s.paneId());
assertTrue(worktrees.removeCalls().isEmpty(),
"a worktree that turned dirty between the pre-stop read and removal must be preserved");
assertEquals(2, worktrees.hasUncommittedCallCount(),
"the fix re-reads hasUncommitted exactly once more, immediately before removal");
}
@Test
void releasePreservesAWorktreeWhoseLateRecheckCannotBeRead() {
// fleetd #316 invariant 1, which no test pinned when the fix landed: the late re-check
// fails toward PRESERVING. Found by mutation — flipping dirtyImmediatelyBeforeRemoval's
// catch from `return true` to `return false` turned the guard into a cause of the very
// data loss it was added to stop, and the whole suite stayed green. The pre-stop read
// succeeds and says clean (call 0); the read that authorises the removal throws (call 1).
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees()
.dirtySequence(false)
.failHasUncommittedOnCall(1, new WorktreeException("git status exited 128"));
SessionManager sessions = sessionManager(herdr, worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-316c", null));
sessions.release(s.paneId());
assertTrue(worktrees.removeCalls().isEmpty(),
"a worktree whose state cannot be read immediately before removal must be kept: "
+ "preserving costs disk, deleting on a guess destroys work with no other copy");
assertEquals(2, worktrees.hasUncommittedCallCount(),
"the late re-check still runs — it is the throwing call, not a skipped one");
}
@Test
void releaseSnapshotsWorkFoundOnlyByTheLateRecheck() {
// #316's second half: the pre-stop dirty=false means trySnapshot never ran for this
// session, so the late-discovered work would otherwise have no refs/wip/* copy at all —
// only the on-disk preserve. The re-check path must snapshot it too.
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees().dirtySequence(false, true);
SessionManager sessions = sessionManager(herdr, worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-316b", null));
sessions.release(s.paneId());
assertEquals(java.util.List.of(s.worktree()), worktrees.snapshotCalls(),
"the newly-dirty worktree is snapshotted even though the pre-stop check saw it clean");
}
@Test
void releaseStillRemovesAWorktreeThatStaysCleanOnTheLateRecheck() {
// The ordinary, non-racing case: nothing else changes behaviour when the second read
// agrees with the first.
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees().dirty(false);
SessionManager sessions = sessionManager(herdr, worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-316c", null));
sessions.release(s.paneId());
assertEquals(java.util.List.of(s.worktree()), worktrees.removeCalls(),
"a worktree that is still clean on the late recheck is removed as before");
}
@Test
void releaseNeverReChecksAWorktreeAlreadyPreservedByTheFirstDirtyCheck() {
// Invariant 4 from #316: no second unconditional git status. A release that already
// decided to preserve (the ordinary CB-576 dirty path) must not pay for a second read.
FakeHerdr herdr = new FakeHerdr();
RecordingWorktrees worktrees = new RecordingWorktrees().dirty(true);
SessionManager sessions = sessionManager(herdr, worktrees);
MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
new WorktreeRequest("cb-316d", null));
sessions.release(s.paneId());
assertEquals(1, worktrees.hasUncommittedCallCount(),
"a release that already preserves on the first read must not re-check before "
+ "skipping the removal it was never going to do");
assertTrue(worktrees.removeCalls().isEmpty());
}
/**
* fleetd #283 defect 1 changed this test's own premise, so its assertions are updated along
* with the production fix. Before #283, the middle session's worktree-removal failure escaped