Compare commits

...

34 Commits

Author SHA1 Message Date
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 53a533afb4 #317: refuse an unresolved caller instead of promoting it to primary
CI / contract (pull_request) Successful in 47s
CI / build (pull_request) Successful in 1m52s
ConnectionIdentity.resolve() called pids.pidForLocalPort(remotePort),
which returns -1 both on a real failure and (silently, no log line)
when lsof just finds no matching process. terminalForPid(-1) then
matches no pane, so CallerResolver's loopback-trust fallback could not
tell that caller apart from a genuine primary and handed it
Principal.primary(...) — granting SPAWN, STOP, SEND and DRAIN to a
worker whose PID lookup failed. This is the escalation PaneLocator's
own javadoc already names; CB-161's ancestry walk only helps once a
candidate pid exists, and a failed lookup has none.

Fix: ConnectionIdentity.Caller gets a resolved() predicate (pid > 0),
centralised next to the -1 sentinel it tests for the same reason
isLoopback() is centralised (fleetd #305: two independent copies of
one rule already drifted once). CallerResolver's loopback-trust
fallback now requires c.resolved() before granting PRIMARY; an
unresolved caller gets Principal.anonymous() — the same already-tested
"authenticated as nothing" outcome used everywhere else in that
method, so the refusal is a clean, named, unsurprising result rather
than something that looks like a bug.

Also logs the previously-silent "lsof ran clean, found no match" case
in LsofPeerPidLookup at DEBUG, since that (not a slow lsof — the
waitFor result was already discarded) is the likelier real trigger.

loopbackTrustTreatsANonWorkerLoopbackCallerAsThePrimary is untouched
and still green: a real pid that owns no pane (the actual primary) is
still resolved() and still PRIMARY. Token mode is unaffected — it
never consults c.pid() at all.

Mutation-tested: reverting only the CallerResolver.java guard
reproduces the escalation exactly (aFailedPeerPidLookupIsRefusedNotPromotedToPrimary
fails with "expected: <ANONYMOUS> but was: <PRIMARY>").
2026-09-04 13:58:14 +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
Dai Ha b9c2cf69f4 Merge #307: a worker's real reply after an ask timeout completes its ticket instead of stranding
CI / build (push) Successful in 2m5s
CI / contract (push) Successful in 2m19s
2026-09-04 13:31:43 +07:00
Dai Ha 8beae50fe7 Merge #308: refuse spawns once the shutdown drain has started, and sweep stragglers
CI / contract (push) Successful in 1m49s
CI / build (push) Successful in 3m0s
2026-09-04 13:28:09 +07:00
Dai Ha b2a58cb966 Merge #309: clean up partial worktree state when git worktree add fails 2026-09-04 13:28:04 +07:00
Dai Ha b8b25cf74c #307: an ask() timeout no longer strands the worker's real reply
CI / contract (pull_request) Successful in 42s
CI / build (pull_request) Successful in 1m30s
MessageService.reply()'s async-recovery path (askAnsweredAsyncTasks)
required a live Task.turnId, but ask()'s own TimeoutException handler
calls clearAsyncQuestion(turnId, true) — deliberately forgetting turnId
so hasAsyncQuestion() stops reporting the target BUSY. That made a
worker's eventual real fleet_reply, after an unanswered fleet_ask, fall
through to the inbox: fleet_poll{ticket} stayed PENDING forever and was
later force-failed with the false reason "session released before it
replied".

Fix: a new Task.askTimedOut marker is set (markAskTimedOut) right
before the turnId is forgotten, and askAnsweredAsyncTasks accepts it in
place of a live turnId. The marker never touches asyncTasksByTurn, so
the BUSY-release behaviour (invariant 1) is untouched. The existing
ambiguity guard (candidates.size() > 1 -> inbox, never guess) still
applies unchanged, but is now genuinely reachable rather than pure
defence in depth, since an ask timeout frees its target for a fresh,
independent delegation — the affected javadocs are updated to say so.

Tests: MessageServiceTest.aReplyAfterAnAskTimeoutStillCompletesTheAsyncTicket
(positive, mutation-proven) and
.twoAskTimedOutTicketsOnOneTargetFallBackToTheInboxRatherThanGuess
(negative/ambiguity). FleetMcpTest's
unansweredAsyncAskReturnsTheTicketToPending was renamed and its final
assertion updated — it had pinned the old (buggy) inbox-stranding
behaviour as expected.
2026-09-04 13:23:01 +07:00
Dai Ha f159ca7d27 #310: log when a reap is skipped because the record changed
CI / contract (push) Successful in 1m7s
CI / build (push) Successful in 1m46s
The compare-and-release declines silently. This race is unobservable by
construction, so a reaper that quietly stops reaping is the hardest kind of
behaviour to diagnose later. One debug line names the pane and the likely
cause.
2026-09-04 13:22:55 +07:00
Dai Ha 83f2aea60f #308: refuse a spawn once the shutdown drain has started, and sweep stragglers
CI / contract (pull_request) Successful in 1m28s
CI / build (pull_request) Successful in 1m36s
drainAll iterated a one-shot registry snapshot with nothing to refuse a new
fleet_spawn while the drain was still running (mcp.close() only runs 8 calls
after sessions.close() in the shutdown hook). A session registered in that
window was never visited by the drain loop: its pane kept running and its
worktree was never preserved, with the in-memory registry gone at exit.

Fix, both mechanisms as the issue asked for (neither alone is complete):

- SessionManager.acquire now checks a `draining` flag, flipped true at the
  very start of drainAll before the registry snapshot is even taken, and
  throws the new ShuttingDownException (invariant 3: fail loudly, say why).
  FleetMcp.spawn and FleetApp.spawnMember surface it as a clean error/503
  rather than an uncaught RuntimeException.
- The flag alone cannot close the whole race: a caller already past the
  check can still be mid-launcher.spawn() (a real herdr round trip) when
  drainAll snapshots the registry. drainAll now re-reads the registry once
  its main pass finishes and drains whatever straggler landed there too,
  bounded by the SAME whole-drain deadline (invariant 1: timeoutNanos stays
  a budget for the whole drain, never extended for a straggler).
- ReleaseCause.SHUTDOWN still preserves worktrees for both the initial pass
  and the sweep (invariant 2, unchanged release() path).

Tests (SessionManagerTest): a guard test proving acquire() throws once
drainAll has started, and a race test using a launcher double that blocks
the second spawn() and the first stop() call to force, deterministically,
the exact interleaving where a spawn passes the guard before drainAll flips
it and only registers after the initial snapshot — proving the post-loop
sweep catches it.

Shape check (SessionManager.java only, not fixed): reapIdle has the same
shape — a decision made from a roster() snapshot, then acted on via
release(s.paneId()) with no re-check of the session's current state.
2026-09-04 13:21:53 +07:00
Dai Ha 002329adb5 #309: clean partial worktrees after add failure
CI / contract (pull_request) Successful in 1m23s
CI / build (pull_request) Successful in 2m2s
2026-09-04 13:18:57 +07:00
Dai Ha a49671ceb9 #310: prevent idle reap from stopping delivered workers
CI / contract (pull_request) Successful in 1m19s
CI / build (pull_request) Successful in 2m1s
2026-09-04 13:17:51 +07:00
Dai Ha 76672ff016 #306: the post-turn phase gets the same unknown-stall escape as a turn
CI / build (push) Successful in 1m43s
CI / contract (push) Successful in 1m50s
Four latches gate delivery in Injector, and only awaitingCompletion had a way
out of a sustained unknown streak. CB-109 added that escape because a worker
stuck in a state herdr cannot classify never produces a working->idle
boundary. The same is true during post-turn housekeeping, but the escape was
never extended there.

awaitingPostTurnPickup and postTurnObserved are both released only on an
injectable sample, so a worker that goes unknown and stays there wedges: the
target is polled forever, every later message to it is blocked by the delivery
gate, and no onTurnFailed fires, so the session sits at DONE and looks healthy.
The counter did not even increment, since ++unknownSinceTurn sits inside the
awaitingCompletion short-circuit.

postTurnPending needs no escape; it is cleared on the line after the listener
call that sets it.

The escape does not set turnFailed. The delegated turn already completed and
its waiter already resolved — what is outstanding is the /clear. Failing the
turn would drive SessionManager.onFailed on a session that genuinely finished.

Not reachable in the live configuration: the path needs lifecycle.clearAfterTurn,
which fleetd.yaml does not set. It becomes reachable as soon as anyone turns
that supported knob on.

Fixes #306
2026-09-04 13:06:29 +07:00
Dai Ha 9379f92c23 #305: one definition of loopback, so a worker cannot become the primary
CI / contract (push) Successful in 52s
CI / build (push) Successful in 1m40s
ConnectionIdentity and CallerResolver each kept their own isLoopback. They
drifted: the identity resolver accepted only 127.0.0.1, the authorization
check accepted all of 127.0.0.0/8.

A caller from 127.0.0.2 therefore had its identity resolution skipped, so it
carried no terminal, and CallerResolver reads a missing terminal as "not a
worker" — which under loopback-trust, the default mode, is the primary. A
worker got spawn, stop, send and drain. The skip also happens before the PID
ancestry walk, so that defence is bypassed too.

Being strict in ConnectionIdentity was not the safe direction. That predicate
decides whether identity is resolved at all, and resolution is what demotes a
worker, so every address it excluded was one where a worker became the lead.

Measured, not assumed: on Linux the whole 127.0.0.0/8 is bound to lo, and
binding a source of 127.0.0.2 on the fleet host succeeds (curl rc=7, the
connect refused rather than the bind). On macOS the source bind fails (rc=45),
so this workstation was never exposed.

The shared predicate also accepts the IPv4-mapped IPv6 form, which neither
copy handled. That one failed in the safe direction: a primary on
::ffff:127.0.0.1 was refused as anonymous.

No transport-level test binds a real 127.0.0.2 source — it cannot run on
macOS. The reasoning is recorded on the issue.

Fixes #305
2026-09-04 12:58:21 +07:00
Dai Ha 21ff63b11d #304: the member routes report a herdr failure instead of a bare 500
CI / build (push) Successful in 1m34s
CI / contract (push) Successful in 1m53s
POST /members and DELETE /members/{paneId} were the two routes in FleetApp
with no catch (HerdrException). FleetApp has no Javalin exception mapper, so
the exception escaped as the default 500 with the body "Server Error" — no
herdr code, no herdr message. fleet_spawn and fleet_stop catch the same
exception and report a named error, so this was the same one-door-guarded
shape as #297.

Both now go through the existing herdrError mapper: 404 when herdr says the
target is gone, 502 otherwise. That is an answer the caller can act on.

stopMember matters more than spawnMember. SessionManager.release deregisters
the session, notifies the release listener and preserves a dirty worktree
before it calls launcher.stop, so a throw from that stop arrives after the
teardown the caller asked for has already happened. A bare 500 told the caller
to retry and carried nothing to explain what went wrong.

The two existing tests that asserted 500 now assert 502 and check the error
body. Neither was about the status code: one guards that a failed teardown is
not reported as a successful 204, the other that a failed spawn still closes
its tab. Both properties are unchanged.

Fixes #304
2026-09-04 12:48:21 +07:00
Dai Ha 21844b54d7 #302: fleet_reply refuses blank content too, matching fleet_send
CI / contract (push) Successful in 1m15s
CI / build (push) Successful in 3m34s
MessageService.reply now throws on blank content. FleetMcp.reply guarded only
against null, and its handler is a bare BiFunction with no try/catch, so a
whitespace-only fleet_reply left the handler as an uncaught
IllegalArgumentException instead of the clean tool error null already got.
fleet_send has always used isBlank here; reply now matches it.

The worker found that null/isBlank difference and reported it as a correction
to my ticket, which had quoted the guard wrongly. It was right: I grepped the
error string and assumed the condition matched its sibling.
2026-09-04 12:36:23 +07:00
Dai Ha 4769481515 Merge #302: a reply with no content is refused, not silently delivered
REST read content with .path("content").asText(""), so a body missing the key
became an empty string that resolved the lead's waiter. The turn completed and
the lead saw a member that finished and reported nothing, indistinguishable
from one that genuinely said nothing. The guard went into MessageService.reply,
which both doors call, rather than being written a second time in FleetApp.
2026-09-04 12:33:47 +07:00
30 changed files with 2579 additions and 118 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).
@@ -236,7 +236,15 @@ public final class CallerResolver {
// loopback-trust: same-host callers that are not workers are the primary. A non-loopback
// caller is anonymous even here — and startup refuses that combination anyway
// (FleetConfig.validateAuthExposure), so this is defence in depth, not the control.
return isLoopback(remoteAddr) ? Principal.primary(c.pid()) : Principal.anonymous();
//
// fleetd #317: "not a worker" must not be conflated with "identity unresolved". The real
// primary is a real process — its pid resolves (c.resolved()), it just owns no herdr pane.
// A caller whose peer-PID lookup failed (LsofPeerPidLookup's -1 sentinel — on any failure,
// silently including "lsof found no match") has no such pid, and PaneLocator's own javadoc
// already names what happens if that case is handed the primary role: a worker→primary
// escalation. So an unresolved caller is refused (ANONYMOUS — the same clean, already-tested
// "authenticated as nothing" outcome used everywhere else in this method), never promoted.
return isLoopback(remoteAddr) && c.resolved() ? Principal.primary(c.pid()) : Principal.anonymous();
}
private boolean presentedTokenMatches(String authorizationHeader) {
@@ -262,11 +270,15 @@ public final class CallerResolver {
return token.isEmpty() ? null : token;
}
/**
* fleetd #305: delegates to {@link ConnectionIdentity#isLoopback}. This used to be a second,
* independent copy of the same rule, and the two drifted: this one accepted all of
* {@code 127.0.0.0/8}, {@code ConnectionIdentity}'s accepted only {@code 127.0.0.1}. A caller
* from {@code 127.0.0.2} therefore had its identity skipped (so it had no terminal) and was
* then read as loopback here — which under loopback-trust is the primary. Sharing the inputs
* would not have prevented that; only sharing the computation does.
*/
private static boolean isLoopback(String remoteAddr) {
if (remoteAddr == null) {
return false;
}
return remoteAddr.equals("127.0.0.1") || remoteAddr.equals("::1")
|| remoteAddr.equals("0:0:0:0:0:0:0:1") || remoteAddr.startsWith("127.");
return ConnectionIdentity.isLoopback(remoteAddr);
}
}
@@ -22,7 +22,7 @@ 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>
@@ -39,29 +39,95 @@ import java.util.function.Supplier;
* 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>Split</strong> (fleetd #330) — read <em>both</em> ways at different sites, so the
* key does not fit any class above as a whole: {@code health:} and {@code coordinator:}.
* 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>
* </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: {@code bind:},
* {@code herdrSocket:}, {@code broker:} and {@code auth:}. The socket is bound, the broker
* connection is open, and the auth mode decides who may reach the port that is already
* listening.</li>
* </ul>
*
* <p><strong>The denominator, measured on 2026-09-04 (fleetd #330).</strong> {@code FleetConfig} has
* 22 top-level record components. Two of them are named nowhere in this file, and the reason is the
* same for both: {@code memberCredentials} and {@code memberLoginShell} are <strong>hot</strong> and
* correctly absent — both are read live off {@code config.get()} 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.
* <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. Twice now — {@code worktreeGroup} (#323) and
* {@code primary}/{@code configReload} (#326) — 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}.
*
* <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 +137,24 @@ 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");
private final Path path;
private final AtomicReference<FleetConfig> current;
@@ -102,25 +182,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 +224,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,8 +276,9 @@ 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;
}
@@ -221,6 +321,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 +388,60 @@ 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). Unlike {@link #changedDeferredKeys}, this does not try to
* tell which sub-field moved: any change to {@code health:} or {@code coordinator:} 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.
*/
private static boolean sameLaunchSettings(FleetConfig.Profile a, FleetConfig.Profile b) {
return Objects.equals(a.baseUrl(), b.baseUrl())
private 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");
}
// 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 two split keys the class doc documents.
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;
}
/**
* {@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 +467,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());
}
}
@@ -157,6 +157,7 @@ public final class Injector {
boolean awaitingCompletion; // a delivered message's turn is not yet known-complete
boolean turnObserved; // saw a real `working` sample since that delivery (turn ran)
int unknownSinceTurn; // consecutive `unknown` samples while a delegation is outstanding (CB-109)
int unknownSincePostTurn; // the same, for the post-turn housekeeping phase (fleetd #306)
int notReadySincePoll; // consecutive injectable samples a queued message waited on the readiness gate (CB-114)
boolean postTurnPending; // completion observed; adapter housekeeping has not started yet
boolean awaitingPostTurnPickup;
@@ -217,10 +218,12 @@ public final class Injector {
t.awaitingPickup = false;
t.injectableSincePickup = 0;
t.unknownSinceTurn = 0;
t.unknownSincePostTurn = 0;
t.notReadySincePoll = 0;
if (t.awaitingCompletion) t.turnObserved = true;
} else if (status.injectable()) { // IDLE or BLOCKED
t.unknownSinceTurn = 0;
t.unknownSincePostTurn = 0;
if (t.awaitingPostTurnPickup) {
if (++t.injectableSincePostTurnPickup >= PICKUP_GRACE_POLLS) {
t.awaitingPostTurnPickup = false;
@@ -315,6 +318,24 @@ public final class Injector {
t.unknownSinceTurn = 0;
turnFailed = true;
}
// fleetd #306: the same escape for the post-turn housekeeping phase. Four latches
// gate delivery (awaitingCompletion, postTurnPending, awaitingPostTurnPickup,
// postTurnObserved) and only the first had a way out of a sustained unknown streak —
// a gate that closed one direction only. The other two below are released here as
// well; postTurnPending needs no escape because it is cleared unconditionally on the
// line after the listener call that sets it.
//
// This does NOT set turnFailed. The delegated turn already completed and its waiter
// already resolved — what is outstanding is adapter housekeeping (the `/clear`).
// Reporting a turn failure here would drive SessionManager.onFailed on a session
// that genuinely finished its work, which is a worse lie than the wedge.
if ((t.awaitingPostTurnPickup || t.postTurnObserved)
&& ++t.unknownSincePostTurn >= TURN_STALL_GRACE_POLLS) {
t.awaitingPostTurnPickup = false;
t.postTurnObserved = false;
t.injectableSincePostTurnPickup = 0;
t.unknownSincePostTurn = 0;
}
}
// Reclaim the entry once the worker is fully quiescent (nothing queued, no pickup or
@@ -36,6 +36,25 @@ public final class ConnectionIdentity {
* primary / an off-host client) and its {@code pid} (or {@code -1} if not resolvable).
*/
public record Caller(String terminal, long pid) {
/**
* Whether the OS peer-PID lookup actually succeeded — {@code false} means {@code pid} is
* the {@code -1} sentinel, not a real process id, so this caller's identity could not be
* established at all. That is a different fact from a real pid that simply owns no worker
* pane (the primary's own connection): the primary is {@code resolved()} and has a
* {@code null terminal}; an unresolvable caller is {@code !resolved()} and also has a
* {@code null terminal}. The two look identical through {@link #terminal} alone, which is
* exactly how fleetd #317 happened — a failed {@code lsof} lookup and a genuine primary both
* fell through to {@code Principal.primary(...)}.
*
* <p>Centralised here, next to the sentinel it tests, for the same reason
* {@link ConnectionIdentity#isLoopback} is centralised rather than left for each caller to
* reimplement: a raw {@code pid > 0} check duplicated at every call site is precisely the
* "one rule, two copies" shape that let #305 drift.
*/
public boolean resolved() {
return pid > 0;
}
}
/** Resolve the caller's terminal and PID from one peer-PID lookup. */
@@ -60,7 +79,44 @@ public final class ConnectionIdentity {
return pid > 0 ? cwds.cwdForPid(pid) : null;
}
private static boolean isLoopback(String addr) {
return "127.0.0.1".equals(addr) || "::1".equals(addr) || "0:0:0:0:0:0:0:1".equals(addr);
/**
* Whether {@code addr} is a same-host address, and therefore one whose peer PID is worth
* looking up. <strong>This is the one definition of loopback in the daemon</strong> —
* {@code CallerResolver} calls it rather than keeping its own, because the two used to differ
* and that difference was a privilege escalation (fleetd #305).
*
* <p>The whole of {@code 127.0.0.0/8} counts, not just {@code 127.0.0.1}. On Linux every
* address in that range is bound to {@code lo} by default, so a process can connect to
* {@code 127.0.0.1:8765} with a source address of {@code 127.0.0.2} — measured on the Linux
* fleet host, where binding that source succeeds.
*
* <p><strong>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) {
return false;
}
String a = addr.startsWith("::ffff:") ? addr.substring(7) : addr; // IPv4-mapped IPv6
return a.startsWith("127.") || "::1".equals(a) || "0:0:0:0:0:0:0:1".equals(a);
}
}
@@ -22,6 +22,7 @@ import dev.ltms.fleet.placement.BackendQuarantine;
import dev.ltms.fleet.placement.PlacementException;
import dev.ltms.fleet.session.SessionManager;
import dev.ltms.fleet.session.MemberSession;
import dev.ltms.fleet.session.ShuttingDownException;
import dev.ltms.fleet.session.WorktreeRequest;
import dev.ltms.fleet.peer.MemberRole;
import dev.ltms.fleet.peer.PeerLauncher;
@@ -813,7 +814,12 @@ public final class FleetMcp {
return error("fleet_reply is for workers only — could not identify the calling worker "
+ "from the connection");
}
if (content == null) {
// fleetd #302: isBlank, not == null, to match fleet_send's own guard above. MessageService
// .reply now REJECTS blank content, and this handler is a bare BiFunction with no try/catch
// around it — so a whitespace-only fleet_reply would leave here as an uncaught
// IllegalArgumentException instead of this clean tool error. Null and whitespace are the
// same mistake by the caller and must get the same answer.
if (isBlank(content)) {
return error("content is required");
}
messages.reply(callerTerminal, content);
@@ -957,6 +963,10 @@ public final class FleetMcp {
return text(json(memberView(member)));
} catch (GuardException e) {
return error("subscription boundary: " + e.getMessage());
} catch (ShuttingDownException e) {
// fleetd #308: the daemon's shutdown drain has already started — refuse loudly rather
// than register a session drainAll will never see again.
return error("shutting down: " + e.getMessage());
} catch (PlacementException e) {
// CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — distinct
// from "profile does not exist" below.
@@ -41,6 +41,14 @@ public final class LsofPeerPidLookup implements PeerPidLookup {
if (!p.waitFor(2, TimeUnit.SECONDS)) {
p.destroyForcibly();
}
if (found < 0) {
// fleetd #317: this is the silent path — lsof ran clean and simply reported no
// matching process (e.g. queried before the OS socket table settles). Previously
// this logged nothing at all, which is exactly why the escalation went unnoticed;
// the exception path below already logs. A caller now refused because of this is
// still refused (never promoted) — this line only makes the refusal diagnosable.
log.debug("lsof peer-pid lookup for port {} found no matching process", port);
}
return found;
} catch (Exception e) {
log.debug("lsof peer-pid lookup for port {} failed: {}", port, e.getMessage());
@@ -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));
}
/**
@@ -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)) {
@@ -187,6 +187,24 @@ public final class MessageService {
private volatile Long completedNanos;
private volatile Reply question;
private volatile String turnId;
/**
* Set when this task's {@code fleet_ask} lapsed with no answer (fleetd #307):
* {@link #clearAsyncQuestion} then forgets {@link #turnId} (nulls it and drops the task from
* {@code asyncTasksByTurn}) so {@link #hasAsyncQuestion} stops reporting the target BUSY — a
* later {@code fleet_send} to it must be accepted, not refused. But the worker's turn is
* still genuinely live: it resumed on its own and will eventually call its real
* {@code fleet_reply}. Losing {@link #turnId} loses {@link #askAnsweredAsyncTasks}' only
* signal that such a reply belongs to this task, so that reply used to fall straight to the
* inbox and strand — {@code fleet_poll} stayed {@code PENDING} forever, later force-failed by
* {@link #abandon} with the misleading "session released before it replied". This flag is a
* second, independent signal that survives the forgetting: {@link #askAnsweredAsyncTasks}
* accepts it in place of a live {@link #turnId}, without ever re-adding the task to
* {@code asyncTasksByTurn} (so the BUSY release is untouched). Cleared implicitly once
* {@link #future} resolves — every match in {@link #askAnsweredAsyncTasks} already requires
* {@code !future.isDone()}, so a task that recovered (or was later failed by
* {@link #abandon}) can never match again regardless of this flag's value.
*/
private volatile boolean askTimedOut;
private Task(String ticket, String target, LongSupplier nowNanos) {
this.ticket = ticket;
@@ -395,18 +413,16 @@ public final class MessageService {
* {@link Rendezvous#resolveQuestion} must keep today's {@code NO_WAITER} behaviour — questions
* are interactive and must never be queued.
*
* <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks}
* cannot actually return more than one entry today (see its own javadoc for why — in short,
* {@link #hasAsyncQuestion} keeps a target BUSY, so no second task can reach this state, for as
* long as an earlier one's {@code turnId} is still stamped). That is an emergent guarantee from
* two other facts, not one this method enforces, so this branch stays in as defence in depth
* rather than being removed as dead code: if it ever weakens, returning whichever candidate a
* {@code ConcurrentHashMap} iteration reaches first would let a genuine reply complete the
* <em>wrong</em> ticket — silently handing the lead something that reads like a correct answer to
* a delegation the worker never touched, which is worse than a failure because the lead acts on
* it. When more than one candidate exists, guessing is not safe: fall back to the inbox exactly
* as the zero-candidate case does, and let {@link #abandon} apply the eventual recovery
* deterministically instead.
* <p><strong>Ambiguous match also falls to the inbox.</strong> {@link #askAnsweredAsyncTasks} can
* return more than one entry — a reachable state, not a hypothetical one (see its own javadoc:
* an {@code fleet_ask} that lapsed with no answer, fleetd #307, frees the target for a completely fresh
* delegation, which can itself go on to ask-and-lapse before the first worker's real reply
* arrives). Returning whichever candidate a {@code ConcurrentHashMap} iteration reaches first
* would let a genuine reply complete the <em>wrong</em> ticket — silently handing the lead
* something that reads like a correct answer to a delegation the worker never touched, which is
* worse than a failure because the lead acts on it. When more than one candidate exists, guessing
* is not safe: fall back to the inbox exactly as the zero-candidate case does, and let
* {@link #abandon} apply the eventual recovery deterministically instead.
*
* <p><strong>{@code content} is required (fleetd #302).</strong> Both doors that reach this
* method must reject a missing/blank reply the same way, so the check lives here rather than in
@@ -434,15 +450,18 @@ public final class MessageService {
count(FleetMetrics.REPLIES, "path", "rendezvous");
return true; // a live send took it — unchanged fast path
}
// #137: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming a
// turn that {@link #answer} already gave up waiting on. answer()'s own bounded wait (the
// primary's fleet_send{turnId} call, capped well under a minute) can time out and close its
// waiter long before the worker — now actually resuming real work — finishes and replies. That
// reply used to have nowhere to land but the session inbox, leaving the async ticket's future
// unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's abandon() forced it
// FAILED with a misleading "session released before it replied" reason, even though the reply
// had, in fact, arrived. Completing the matching ticket directly here means fleet_poll{ticket}
// sees the real reply instead.
// #137/fleetd #307: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming
// a turn that either answer() (#137) or ask() (fleetd #307) already gave up waiting on:
// - answer()'s own bounded wait (the primary's fleet_send{turnId} call, capped well under a
// minute) can time out and close its waiter long before the worker — now actually resuming
// real work — finishes and replies.
// - ask()'s own wait for the primary can time out first, with the worker resuming on its own
// and finishing unanswered.
// Either way that reply used to have nowhere to land but the session inbox, leaving the async
// ticket's future unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's
// abandon() forced it FAILED with a misleading "session released before it replied" reason,
// even though the reply had, in fact, arrived. Completing the matching ticket directly here
// means fleet_poll{ticket} sees the real reply instead.
List<Task> candidates = askAnsweredAsyncTasks(session);
if (candidates.size() == 1) {
Task orphan = candidates.get(0);
@@ -473,34 +492,42 @@ public final class MessageService {
}
/**
* Every still-open async task on {@code target} whose {@code fleet_ask} was already answered —
* its {@link Task#turnId} is stamped but its {@link Task#question} was cleared by {@link #answer}
* — yet whose future is not resolved yet (#137). Empty if no such task exists, including the
* common case where {@code target}'s worker never used {@code fleet_ask} at all (a task that was
* never asked has {@code turnId == null}, so it can never match here and only ever completes
* through the ordinary rendezvous fast path in {@link #reply}).
* Every still-open async task on {@code target} whose worker is genuinely expected to send a
* real {@code fleet_reply} next with nothing left registered to catch it: either its
* {@code fleet_ask} was already answered — {@link Task#turnId} is stamped but {@link
* Task#question} was cleared by {@link #answer} — or its {@code fleet_ask} lapsed unanswered and
* {@link Task#askTimedOut} marks that (fleetd #307; {@link Task#turnId} is {@code null} by then, forgotten
* so the target is not left BUSY — see {@link Task#askTimedOut}'s own javadoc). Either way the
* task's future is not resolved yet. Empty if no such task exists, including the common case
* where {@code target}'s worker never used {@code fleet_ask} at all (a task that was never asked
* has both {@code turnId == null} and {@code askTimedOut == false}, so it can never match here and
* only ever completes through the ordinary rendezvous fast path in {@link #reply}).
*
* <p><strong>Returns at most one entry today — verified, not assumed.</strong> {@link #send}
* refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is true, and that
* check matches ANY task whose {@code turnId} is still stamped in {@code asyncTasksByTurn} —
* not only while its question is still open. {@link #answer} deliberately leaves that stamp in
* place ({@code clearAsyncQuestion(turnId, false)}) until the resumed turn's own future actually
* resolves, at which point {@link #finishAsyncTask} both removes the stamp AND completes that
* task's future in the same call. So a second task can never reach "{@code turnId} stamped, future
* still open" — the exact pair this method matches on — while a first one already holds it: by
* the time the stamp is gone, so is the eligibility. This is an emergent property of those two
* facts holding together, not something this method (or its callers) enforces on its own — flip
* {@code forgetTurn} to {@code true} in that one {@link #answer} call and it silently stops being
* true, with nothing left to fail loudly. The callers below still handle "more than one" as
* defence in depth against exactly that, not because they exercise it today: {@link #reply}
* treats it as unresolvable and falls back to the inbox; {@link #abandon} would pick the oldest
* deterministically (its own {@code matching} list has no such guarantee — see its javadoc).
* <p><strong>Can return more than one entry — reachable, not just defence in depth.</strong>
* {@link #send} refuses to open a waiter on {@code target} while {@link #hasAsyncQuestion} is
* true, and that check matches ANY task whose {@code turnId} is still stamped in
* {@code asyncTasksByTurn}. While a task's {@code turnId} stays stamped — {@link #answer} leaves
* it in place ({@code clearAsyncQuestion(turnId, false)}) until {@link #finishAsyncTask} removes
* the stamp and completes the future in the same call — no second task on the same target can
* reach an eligible state, because {@link #send} would refuse it as BUSY first. That single-task
* guarantee holds only for the {@code turnId}-stamped half of this method's match: an
* {@link Task#askTimedOut} task is, by construction, no longer stamped in {@code asyncTasksByTurn}
* (that is the whole point of forgetting {@code turnId} in {@link #clearAsyncQuestion}), so the
* target is free the moment one ask lapses. A fresh, independent {@code sendAsync} to the same
* target can then be dispatched, itself pause on {@code fleet_ask}, and itself time out — landing
* a second {@code askTimedOut} task on the very target the first one is still waiting to answer
* for. Two (or more) genuinely open tasks on one target is therefore a real, reachable state
* today, not a hypothetical: {@link #reply} treats it as unresolvable and falls back to the
* inbox rather than guess which task a reply belongs to (guessing wrong would hand the lead a
* plausible-looking answer to a delegation the worker never touched — worse than a failure,
* because the lead acts on it); {@link #abandon} instead picks the oldest deterministically (its
* own {@code matching} list has a different, wider match — see its javadoc).
*/
private List<Task> askAnsweredAsyncTasks(String target) {
List<Task> candidates = new ArrayList<>();
for (Task task : tasks.values()) {
if (target.equals(task.target) && task.question == null && task.turnId != null
&& !task.future.isDone()) {
if (target.equals(task.target) && task.question == null && !task.future.isDone()
&& (task.turnId != null || task.askTimedOut)) {
candidates.add(task);
}
}
@@ -898,6 +925,12 @@ public final class MessageService {
return new AskResult(AskOutcome.ANSWERED, answer);
} catch (TimeoutException e) {
log.debug("fleet_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
// fleetd #307: mark the task BEFORE clearAsyncQuestion(forgetTurn=true) below drops it out of
// asyncTasksByTurn and nulls its turnId — that forgetting is deliberate and stays (it is
// what keeps the target from staying BUSY forever), but it would otherwise also erase
// askAnsweredAsyncTasks' only signal that the worker's eventual real fleet_reply still
// belongs to this task, stranding it in the inbox with a false "never replied" verdict.
markAskTimedOut(ticket.turnId());
clearAsyncQuestion(ticket.turnId(), true);
return new AskResult(AskOutcome.TIMED_OUT, null);
} catch (ExecutionException e) {
@@ -1162,6 +1195,23 @@ public final class MessageService {
return task;
}
/**
* Mark {@code turnId}'s task as having a {@code fleet_ask} that lapsed with no answer (fleetd #307), so
* {@link #askAnsweredAsyncTasks} still recognizes the worker's eventual real {@code fleet_reply}
* as belonging to it after {@link #clearAsyncQuestion}'s {@code forgetTurn=true} erases
* {@link Task#turnId} — see {@link Task#askTimedOut}. Must be called before that forgetting, while
* {@code turnId} can still resolve the task in {@code asyncTasksByTurn}; a lookup afterward would
* find nothing. Only when it matches the task's current turn — same guard as
* {@link #clearAsyncQuestion} — so a chained second {@code fleet_ask} (#282) that already moved
* the task to a fresh {@code turnId} cannot mark it for a turn that is no longer its own.
*/
private void markAskTimedOut(String turnId) {
Task task = asyncTasksByTurn.get(turnId);
if (task != null && turnId.equals(task.turnId)) {
task.askTimedOut = true;
}
}
/** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
// CB-582: tell the push loop first — like ticketCollected, a removal for a turnId it never
@@ -1180,14 +1230,61 @@ 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);
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);
}
}
/**
* 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;
}
/**
* 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);
}
/** Complete the async ticket correlated to a specific answered turn. */
private void finishAsyncTask(String turnId, Reply result) {
Task task = asyncTasksByTurn.get(turnId);
@@ -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");
}
@@ -19,6 +19,7 @@ import dev.ltms.fleet.peer.PeerUnreachableException;
import dev.ltms.fleet.placement.PlacementException;
import dev.ltms.fleet.msg.MessageService;
import dev.ltms.fleet.session.SessionManager;
import dev.ltms.fleet.session.ShuttingDownException;
import dev.ltms.fleet.peer.MemberRole;
import dev.ltms.fleet.session.MemberSession;
import dev.ltms.fleet.session.WorktreeRequest;
@@ -501,6 +502,11 @@ public final class FleetApp {
ctx.status(201).json(view(member));
} catch (GuardException e) {
ctx.status(403).json(Map.of("error", "subscription_boundary", "detail", e.getMessage()));
} catch (ShuttingDownException e) {
// fleetd #308: the daemon's shutdown drain has already started — 503, not a bare 500,
// so this reads the same as PlacementException below: valid request, refused because
// of a transient daemon state rather than a bad argument.
ctx.status(503).json(Map.of("error", "shutting_down", "detail", e.getMessage()));
} catch (PlacementException e) {
// CB-599: no candidate had capacity (maxLoad, quarantine, or all-exhausted) — a benign,
// likely-transient refusal, distinct from "profile does not exist" below. 503: the
@@ -510,6 +516,12 @@ public final class FleetApp {
ctx.status(400).json(Map.of("error", "unknown_profile", "detail", e.getMessage()));
} catch (PeerUnreachableException e) {
ctx.status(502).json(Map.of("error", "spawn_timeout", "detail", e.getMessage()));
} catch (HerdrException e) {
// fleetd #304: not every herdr failure on the spawn path is a readiness timeout, so
// PeerUnreachableException above does not cover this. Without this catch the exception
// escapes to Javalin's default 500, while fleet_spawn reports the same failure as a
// clean named error (FleetMcp.spawn) — the #297 one-door-guarded shape.
herdrError(ctx, e);
}
}
@@ -530,13 +542,29 @@ public final class FleetApp {
return (s == null || s.isBlank()) ? null : s;
}
/** Tear a worker down by pane id. */
/**
* Tear a worker down by pane id.
*
* <p>fleetd #304: the {@code HerdrException} catch is not cosmetic. {@code release} deregisters
* the session, notifies the release listener and preserves a dirty worktree <em>before</em> it
* calls {@code launcher.stop}, so a throw from that stop arrives after the teardown the caller
* asked for has already happened. Letting it escape gave Javalin's default 500, which tells the
* caller to retry — and the retry finds nothing in the registry, reaches the same stop, and
* throws again, so it can never succeed. {@code herdrError} instead answers 404 ("the pane is
* gone, stop retrying") or 502 ("herdr is upstream and broken, a retry may help"), matching what
* {@code fleet_stop} reports for the same failure.
*/
private void stopMember(Context ctx) {
String paneId = ctx.pathParam("paneId");
if (!allow(ctx, routeAction("DELETE /members/{paneId}"), paneId)) {
return;
}
sessions.release(paneId);
try {
sessions.release(paneId);
} catch (HerdrException e) {
herdrError(ctx, e);
return;
}
ctx.status(204);
}
@@ -93,6 +93,9 @@ public final class GitWorktrees implements Worktrees {
/** OS group name for {@link #shareWithGroup} (fleetd #185 stage 3); {@code null} ⇒ feature off. */
private final String group;
private final Consumer<String> afterWorktreeAdded;
/** How the initial {@code git worktree add} command runs. Package-private test seam for an
* interrupted command after Git has made worktree state. */
private final Function<String[], String> worktreeAddRunner;
/** How {@link #shareWithGroup}'s processes (git config / chgrp / chmod / find) actually run.
* Defaults to the real {@link #exec(String...)}. Package-private test seam so a unit test can
* prove "no group configured ⇒ zero processes spawned" and inspect exactly what a configured
@@ -132,7 +135,13 @@ public final class GitWorktrees implements Worktrees {
/** Test seam combining a configurable {@code group} with {@link #afterWorktreeAdded}. */
GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded) {
this(configuredRoot, group, afterWorktreeAdded, null);
this(configuredRoot, group, afterWorktreeAdded, null, null);
}
/** Test seam for changing how {@link #shareWithGroup}'s processes run. */
GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded,
Function<String[], String> shareGroupRunner) {
this(configuredRoot, group, afterWorktreeAdded, shareGroupRunner, null);
}
/**
@@ -141,13 +150,15 @@ public final class GitWorktrees implements Worktrees {
* exactly what commands a configured group runs, without a real second OS user/group.
*
* @param shareGroupRunner {@code null} ⇒ the real {@link #exec(String...)}.
* @param worktreeAddRunner {@code null} ⇒ the real {@link #exec(String...)}.
*/
GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded,
Function<String[], String> shareGroupRunner) {
Function<String[], String> shareGroupRunner, Function<String[], String> worktreeAddRunner) {
this.configuredRoot = configuredRoot;
this.group = (group == null || group.isBlank()) ? null : group;
this.afterWorktreeAdded = afterWorktreeAdded == null ? _ -> {} : afterWorktreeAdded;
this.shareGroupRunner = shareGroupRunner != null ? shareGroupRunner : this::exec;
this.worktreeAddRunner = worktreeAddRunner != null ? worktreeAddRunner : this::exec;
}
@Override
@@ -166,8 +177,8 @@ public final class GitWorktrees implements Worktrees {
String wt = path.toAbsolutePath().toString();
log.info("adding worktree branch={} path={} base={}", branch, wt, base);
removeUserInfoFromHttpsOrigin(repoRoot);
exec("git", "-C", repoRoot, "worktree", "add", wt, "-b", branch, base);
try {
worktreeAddRunner.apply(new String[] {"git", "-C", repoRoot, "worktree", "add", wt, "-b", branch, base});
afterWorktreeAdded.accept(wt);
requireCredentialFreeHttpsOrigin(wt);
configureEnvironmentCredentialHelper(repoRoot, wt);
@@ -181,10 +192,11 @@ public final class GitWorktrees implements Worktrees {
}
/**
* {@code add()} has already created the worktree and its branch by the time any step from
* {@link #afterWorktreeAdded} through {@link #isolateToolSurface} can throw — including
* {@code git worktree add} may have created the worktree and its branch by the time it, or any
* later step through {@link #isolateToolSurface}, throws. This includes
* {@link #requireCredentialFreeHttpsOrigin}, an intended security refusal, not only an IO
* accident. Without this, {@code add()} never returns, so its caller
* accident. A Git-reported {@code worktree add} failure usually creates nothing, but an
* interrupted command can leave partial state. Without cleanup, {@code add()} never returns, so its caller
* ({@code SessionManager#acquireWithWorktree}) never receives a path to register or clean up:
* its local {@code path} stays null, the {@code if (path != null)} guard in its own catch block
* never runs, and the worktree directory and branch leak on disk forever with nothing tracking
@@ -204,6 +216,10 @@ public final class GitWorktrees implements Worktrees {
* used in {@code SessionManager#acquireWithWorktree}'s own catch block.
*/
private void cleanupAfterAddFailure(String repoRoot, String worktreePath, String branch, RuntimeException original) {
if (!Files.exists(Path.of(worktreePath))) {
log.debug("provisioning failed before worktree {} existed; nothing to clean up", worktreePath);
return;
}
log.warn("provisioning failed for branch={} path={}: {} — cleaning up before rethrowing",
branch, worktreePath, original.getMessage());
try {
@@ -21,6 +21,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.LongSupplier;
@@ -61,6 +62,8 @@ public final class SessionManager implements TurnListener {
private final LongSupplier nowNanos;
private final int contextCap;
private final boolean clearAfterTurn;
/** Null in production; test seam for the interval before an idle session's conditional release. */
private final Consumer<MemberSession> beforeIdleRelease;
private volatile MemberLifecycle memberLifecycle = MemberLifecycle.NONE;
/**
* CB-586: the repo root the fleet actually works in, remembered the first time a worktree
@@ -71,6 +74,16 @@ public final class SessionManager implements TurnListener {
*/
private volatile String fleetRepoRoot;
/**
* fleetd #308: flips true the instant {@link #drainAll} starts, before its registry snapshot
* is even taken — so a spawn already in flight sees the refusal as early as a plain flag can
* make it. This alone cannot close the race completely: a caller that read {@code false} just
* before the flip can still land in the registry after the snapshot. {@link #drainAll}'s
* post-loop sweep is what catches that straggler; the two mechanisms are deliberately paired,
* see {@link #drainAll}'s javadoc.
*/
private final AtomicBoolean draining = new AtomicBoolean(false);
/** CB-520: notified with a terminalId on every acquire; no-op until wired. */
private final List<Consumer<String>> acquireListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
/** CB-516: notified with a {@link ReleaseDetail} on every release; no-op until wired. */
@@ -102,13 +115,23 @@ public final class SessionManager implements TurnListener {
}
public SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
int contextCap, boolean clearAfterTurn) {
int contextCap, boolean clearAfterTurn) {
this(launcher, worktrees, nowNanos, contextCap, clearAfterTurn, null);
}
/**
* Package-private constructor for a deterministic reap/delivery race test. Production callers
* use the constructor above, whose null hook adds no callback or lock to an ordinary reap.
*/
SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
int contextCap, boolean clearAfterTurn, Consumer<MemberSession> beforeIdleRelease) {
this.launcher = launcher;
this.worktrees = worktrees;
this.presence = new PresenceFleet(this);
this.nowNanos = nowNanos;
this.contextCap = contextCap;
this.clearAfterTurn = clearAfterTurn;
this.beforeIdleRelease = beforeIdleRelease;
}
/**
@@ -181,6 +204,14 @@ public final class SessionManager implements TurnListener {
public MemberSession acquire(String profile, MemberRole role, String requestedCwd, String callerCwd,
String ownerTerminal, WorktreeRequest wt,
String sessionName, String resumeSessionId) {
// fleetd #308: refuse before anything else runs — no slot reservation, no launcher spawn —
// so a caller learns the daemon is going down instead of getting a session drainAll will
// never see again. Checked here because every other acquire(...) overload delegates to
// this one, so this is the single point every spawn path passes through.
if (draining.get()) {
throw new ShuttingDownException("fleetd is shutting down; refusing to spawn a session "
+ "the shutdown drain would never see");
}
MemberRole memberRole = (role == null) ? MemberRole.DEV : role;
requireResumeCapability(profile, resumeSessionId);
// CB-619 / fleetd #123: an explicit profile bypasses placement (CompositePeerLauncher only
@@ -272,10 +303,32 @@ public final class SessionManager implements TurnListener {
*/
private void release(String paneId, ReleaseCause cause) {
MemberSession removed = registry.remove(paneId);
releaseRemoved(paneId, removed, handles.remove(paneId), cause);
}
/**
* Tear a session down only while {@code expected} is still its registry value. A lifecycle
* transition replaces the immutable record, so this prevents a reap based on an old READY or
* DONE record from stopping a worker that delivery has made BUSY.
*/
private boolean releaseIfCurrent(MemberSession expected, ReleaseCause cause) {
if (!registry.remove(expected.paneId(), expected)) {
// A lifecycle transition replaced the record between the caller's check and this remove.
// Log it: this race is by definition unobservable otherwise, and a reaper that silently
// declines to reap is the hardest kind of behaviour to diagnose after the fact.
log.debug("skipping reap of pane={}: its registry record changed after the idle check "
+ "(most likely a delivery made it BUSY)", expected.paneId());
return false;
}
releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
return true;
}
private void releaseRemoved(String paneId, MemberSession removed, PeerHandle removedHandle,
ReleaseCause cause) {
// fleetd #209: remove right alongside the registry entry so a released session's handle is
// never leaked — but keep the local reference below, so the id can still be resolved for
// the ReleaseDetail this teardown notifies with.
PeerHandle removedHandle = handles.remove(paneId);
boolean preserveWorktree = cause == ReleaseCause.SHUTDOWN;
String snapshotRef = null;
if (removed != null) {
@@ -333,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
@@ -351,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
@@ -846,14 +942,18 @@ public final class SessionManager implements TurnListener {
}
long idleNanos = now - s.lastActivityAtNanos();
if (idleNanos > idleTtlNanos) {
log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
// CB-581: one session that fails to release must not abort the whole reaping pass —
// match drainAll's per-session try/catch so the rest of the roster still gets reaped.
try {
release(s.paneId());
reaped++;
if (beforeIdleRelease != null) {
beforeIdleRelease.accept(s);
}
if (releaseIfCurrent(s, ReleaseCause.COMPLETED)) {
log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
reaped++;
}
} catch (RuntimeException e) {
log.warn("reap failed for pane={} terminal={} worktree={}; continuing with "
+ "remaining sessions", s.paneId(), s.terminalId(), s.worktree(), e);
@@ -884,10 +984,40 @@ public final class SessionManager implements TurnListener {
* a reason to delete a worker's only copy of its uncommitted work. A session still {@code BUSY}
* when the timeout expired is abandoned mid-turn and logged loudly so an operator can find its
* kept worktree.
*
* <p>fleetd #308: {@code roster()} is a one-shot snapshot (see its javadoc), and nothing used
* to stop a new session from registering after it was taken — {@link #acquire} stayed open for
* as long as this drain waited on a {@code BUSY} session, up to the whole {@code timeoutNanos}
* budget. Two things close that window, deliberately paired because neither alone is complete:
* {@link #draining} is flipped true before the snapshot is even taken, so {@link #acquire}
* refuses (invariant 3: loudly, via {@link ShuttingDownException}) as much of the window as a
* plain flag can close; and the sweep below re-reads the registry once the initial snapshot has
* fully drained and drains whatever a straggler — a caller that read the flag as {@code false}
* a moment before it flipped — still managed to register. The sweep shares the same
* {@code deadline} rather than getting its own: {@code timeoutNanos} is a budget for the WHOLE
* drain (see above), and a straggler must not buy the drain more time than the flag it lost the
* race against would have. In the ordinary case the sweep finds nothing and costs one empty
* {@link #roster()} call.
*/
void drainAll(long timeoutNanos) {
long deadline = System.nanoTime() + timeoutNanos;
for (MemberSession s : roster()) {
draining.set(true);
drainSnapshot(roster(), deadline);
List<MemberSession> stragglers = roster();
if (!stragglers.isEmpty()) {
log.warn("drain sweep found {} session(s) registered after the drain snapshot was "
+ "taken (raced past the shutdown guard); draining them too", stragglers.size());
drainSnapshot(stragglers, deadline);
}
}
/**
* Drain exactly the sessions in {@code snapshot}, waiting out a {@code BUSY} one against the
* shared whole-drain {@code deadline} before releasing it. Shared by {@link #drainAll}'s main
* pass and its post-loop straggler sweep (fleetd #308) so both honor the same one budget.
*/
private void drainSnapshot(List<MemberSession> snapshot, long deadline) {
for (MemberSession s : snapshot) {
try {
if (s.state() == MemberSession.State.BUSY) {
while (System.nanoTime() < deadline) {
@@ -0,0 +1,18 @@
package dev.ltms.fleet.session;
/**
* Thrown by {@link SessionManager#acquire} when a spawn is requested after the daemon's shutdown
* drain has already begun (fleetd #308).
*
* <p>{@link SessionManager#drainAll} snapshots the registry once and tears down exactly what is
* in that snapshot. A session registered after the snapshot is invisible to the drain loop: its
* pane is left running and its worktree is never preserved, and nothing else ever reclaims
* either — the daemon's in-memory registry dies with the process. Refusing the spawn here,
* loudly, is what stops that session from ever being created in the first place, rather than
* silently handing the caller a session the daemon can no longer manage.
*/
public final class ShuttingDownException extends RuntimeException {
public ShuttingDownException(String message) {
super(message);
}
}
@@ -103,6 +103,54 @@ class CallerResolverTest {
assertEquals(Role.PRIMARY, p.role(), "the historical behaviour, now an explicit choice");
}
// ── fleetd #317: an unresolvable caller must never be promoted to the primary ──────────────────
// #305 closed the trigger where a resolved pid matched no pane *and* had no ancestry walk to
// save it. This is the other trigger PaneLocator's javadoc names: the pid never resolves at
// all — LsofPeerPidLookup returns -1 on any failure, including (silently) "lsof found no
// match" — so there is no candidate pid for an ancestry walk to even attempt.
/**
* The failing-without-the-fix case. Before #317's fix, {@code c.terminal() == null} was the
* only test in the loopback-trust fallback, and an unresolved pid produces exactly that same
* {@code null} terminal as a genuine primary — so this caller was handed
* {@code Principal.primary(...)}, a real worker's failed lookup becoming indistinguishable from
* the lead.
*/
@Test
void aFailedPeerPidLookupIsRefusedNotPromotedToPrimary() {
ConnectionIdentity unresolved = new ConnectionIdentity(new PaneLocator(herdr), _ -> -1);
Principal p = new CallerResolver(unresolved).resolve("127.0.0.1", 55555, null);
assertEquals(Role.ANONYMOUS, p.role(),
"an unresolvable caller must never be silently promoted to the primary");
}
/**
* The companion invariant #317 must not break: a caller whose lookup genuinely succeeded, and
* who simply owns no herdr pane — the real primary's own connection — is still the primary.
* This is {@link #loopbackTrustTreatsANonWorkerLoopbackCallerAsThePrimary} pinned again here,
* named for #317 and placed next to the test it must be distinguished from: same {@code null}
* terminal, opposite verdict, because {@code Caller.resolved()} tells them apart.
*/
@Test
void aRealPidThatOwnsNoPaneIsStillThePrimaryNotRefused() {
Principal p = new CallerResolver(nonWorkerIdentity()).resolve("127.0.0.1", 55555, null);
assertEquals(Role.PRIMARY, p.role());
}
/** #317 point 4: token mode never consults {@code c.pid()}, so a failed lookup must not change it. */
@Test
void tokenModeIsUndisturbedByAnUnresolvedLookup() {
ConnectionIdentity unresolved = new ConnectionIdentity(new PaneLocator(herdr), _ -> -1);
CallerResolver r = new CallerResolver(unresolved, true, "s3cret");
assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 55555, null).role(),
"no credential is still just ANONYMOUS, as before #317 — unchanged by the lookup failing");
assertEquals(Role.PRIMARY, r.resolve("127.0.0.1", 55555, "Bearer s3cret").role(),
"a valid token still authenticates the primary even though the peer-pid lookup failed");
}
@Test
void tokenModeRefusesANonWorkerCallerThatPresentsNoToken() {
Principal p = new CallerResolver(nonWorkerIdentity(), true, "s3cret")
@@ -413,4 +461,29 @@ class CallerResolverTest {
assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, null));
assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, " "));
}
@Test
void aWorkerOnAnyLoopbackSourceAddressIsStillAWorkerNotThePrimary() {
// fleetd #305: the escalation. ConnectionIdentity used to accept only 127.0.0.1, so a
// worker connecting from 127.0.0.2 resolved to no terminal, and this resolver's own
// (wider) loopback check then made it the PRIMARY — granting spawn, stop, send and drain.
// Measured on the Linux fleet host: binding a source of 127.0.0.2 succeeds there, so the
// path is real and not theoretical.
CallerResolver r = new CallerResolver(workerIdentity(), false, null);
for (String src : new String[]{"127.0.0.1", "127.0.0.2", "127.1.2.3", "::ffff:127.0.0.2"}) {
Principal p = r.resolve(src, 55555, null);
assertEquals(Role.WORKER, p.role(), "a worker must stay a worker from source " + src);
assertEquals("term_a", p.terminal(), "worker terminal from source " + src);
}
}
@Test
void aNonWorkerOnAnyLoopbackSourceAddressIsStillThePrimary() {
// The other direction of the same fix: widening the identity check must not demote a
// legitimate same-host primary that happens to connect from another 127.* address.
CallerResolver r = new CallerResolver(nonWorkerIdentity(), false, null);
for (String src : new String[]{"127.0.0.1", "127.0.0.2", "::ffff:127.0.0.1"}) {
assertEquals(Role.PRIMARY, r.resolve(src, 55555, null).role(), "source " + src);
}
}
}
@@ -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,291 @@ 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());
}
/**
* 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,167 @@
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
* — verified by reading that method as of fleetd #330, not derived from this test.
* {@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 = Set.of(
"guard", "worktreeRoot", "worktreeGroup", "primary", "configReload",
"leadHeartbeat", "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs",
"quarantineCooldownSeconds", "profiles");
/**
* 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 fleet} — role pools, {@code charters} and {@code tabLabel} are read live through
* the supplier on {@code CompositePeerLauncher} (class doc, Hot bullet). The one
* documented exception, {@code fleet.leaders}, is read only at startup and genuinely needs
* a restart — a real gap in {@code changedDeferredKeys}, but one the class doc already
* carries and that fleetd #330 explicitly did not re-open (its "seven readers" fact-find
* named {@code health}/{@code coordinator} as the complete set of split-shaped keys, not
* {@code fleet}). Reported as a caveat, not fixed here.</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>
*/
private static final Set<String> HOT_EXCLUDED_TOP_LEVEL_KEYS =
Set.of("placement", "fleet", "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", "fleet", "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");
}
}
@@ -297,6 +297,68 @@ class InjectorTest {
assertTrue(inj.activeTargets().isEmpty(), "the wedged target is reclaimed, not polled forever");
}
/** A listener whose post-turn housekeeping always starts, as SessionManager's does with clearAfterTurn on. */
private static final class PostTurnListener implements TurnListener {
@Override public void onTurnComplete(String target) { }
@Override public boolean hasPostTurnAction(String target) { return true; }
@Override public boolean onTurnCompleteWithPostAction(String target) { return true; }
}
@Test
void aWorkerThatWedgesInUnknownAwaitingPostTurnPickupIsReleased() {
// fleetd #306: the post-turn phase had no way out of a sustained unknown streak, so the
// pickup latch stayed set, the target was polled forever, and every later message to it was
// blocked by the delivery gate — while the session still looked healthy.
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.onStatus(T, AgentStatus.IDLE); // deliver
inj.onStatus(T, AgentStatus.WORKING); // turn starts
inj.onStatus(T, AgentStatus.IDLE); // turn completes; housekeeping dispatched
for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // then wedges
assertTrue(inj.activeTargets().isEmpty(),
"a target wedged awaiting post-turn pickup must be reclaimed, not polled forever");
assertEquals(List.of(), cap.failed,
"the delegated turn already completed — a stuck /clear must not be reported as a failed turn");
}
@Test
void aWorkerThatWedgesInUnknownAfterPickingUpTheResetIsReleased() {
// The sibling latch. postTurnObserved is set when the reset is seen picked up (WORKING) and
// is cleared only on a later injectable sample, so a wedge right after pickup sticks too.
Captor cap = new Captor();
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
inj.enqueue(T, "task", TestTurnTokens.inert(T));
inj.onStatus(T, AgentStatus.IDLE);
inj.onStatus(T, AgentStatus.WORKING);
inj.onStatus(T, AgentStatus.IDLE); // turn complete; reset dispatched
inj.onStatus(T, AgentStatus.WORKING); // reset picked up -> postTurnObserved
for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN);
assertTrue(inj.activeTargets().isEmpty(), "a wedge after reset pickup must also be reclaimed");
assertEquals(List.of(), cap.failed, "still not a turn failure");
}
@Test
void aBriefUnknownDuringPostTurnHousekeepingDoesNotDropTheLatch() {
// The other direction: the escape must not fire on a glitch, or the queued next delegation
// would overtake housekeeping that is still running.
Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener());
inj.enqueue(T, "first", TestTurnTokens.inert(T));
inj.enqueue(T, "second", TestTurnTokens.inert(T));
inj.onStatus(T, AgentStatus.IDLE);
inj.onStatus(T, AgentStatus.WORKING);
inj.onStatus(T, AgentStatus.IDLE); // first completes; reset dispatched
for (int i = 0; i < 10; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // well under the grace
assertFalse(inj.activeTargets().isEmpty(), "a brief glitch must not release the post-turn latch");
assertEquals(List.of("first"), sent(), "the queued delegation must not overtake housekeeping");
}
@Test
void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
Captor cap = new Captor();
@@ -20,6 +20,17 @@ class ConnectionIdentityTest {
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.0.0.1", 55555));
}
@Test
void resolvesWorkerFromAnyLoopbackSourceAddressNotJust127001() {
// fleetd #305. On Linux the whole 127.0.0.0/8 is bound to lo, so a worker can connect with
// a source address of 127.0.0.2. If identity resolution skips that address the caller has
// no terminal, and a caller with no terminal is the primary under loopback-trust — so this
// must resolve the worker, not null.
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.0.0.2", 55555));
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("127.1.2.3", 55555));
assertEquals("term_a", with(_ -> FakeHerdr.WORKER_PID).callerTerminal("::ffff:127.0.0.2", 55555));
}
@Test
void nullForOffHostCaller() {
// A non-loopback peer can't be an on-host worker → treat as primary/unknown.
@@ -32,6 +43,22 @@ class ConnectionIdentityTest {
assertNull(with(_ -> 999_999).callerTerminal("127.0.0.1", 55555));
}
@Test
void callerIsUnresolvedWhenThePeerPidLookupFails() {
// fleetd #317: LsofPeerPidLookup returns -1 on any failure — a fork error, or (silently)
// simply no matching lsof line. Caller.resolved() is the one place that sentinel is tested.
ConnectionIdentity.Caller c = with(_ -> -1).resolve("127.0.0.1", 55555);
assertFalse(c.resolved(), "a -1 pid means the lookup failed, not that this pid owns no pane");
}
@Test
void callerIsResolvedWhenThePidIsRealEvenThoughItOwnsNoPane() {
// The primary's own connection: a real, lsof-found pid that just isn't a worker pane. This
// must read as "resolved" — the distinction #317 turns on.
ConnectionIdentity.Caller c = with(_ -> 999_999).resolve("127.0.0.1", 55555);
assertTrue(c.resolved());
}
@Test
void resolvesTheCallersPidAndCwd() {
// CB-112: the primary maps to no pane, but its PID and cwd are still readable.
@@ -189,7 +189,7 @@ class FleetMcpTest {
}
@Test
void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
void unansweredAsyncAskReturnsTheTicketToPendingThenAWorkersLateReplyStillCompletesIt() throws Exception {
McpSchema.CallToolResult accepted = FleetMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
@@ -203,8 +203,15 @@ class FleetMcpTest {
assertTrue(textOf(ask).contains("no answer"), textOf(ask));
assertTrue(textOf(FleetMcp.poll(messages, ticket, null)).startsWith("[pending"));
// fleetd #307: the worker resumed on its own after the primary never answered, and its real
// fleet_reply must complete its OWN async ticket — not strand in the inbox with
// fleet_poll{ticket} stuck PENDING forever and later force-failed with a false "session
// released before it replied" reason. This used to land in the inbox instead (see the old
// assertion this replaced: messages.drainReplies("term_a").getFirst()...) — that was the bug.
FleetMcp.reply(messages, "term_a", "finished after timeout");
assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
assertEquals("finished after timeout", textOf(FleetMcp.poll(messages, ticket, null)));
assertTrue(messages.drainReplies("term_a").isEmpty(),
"the reply completed its own ticket directly and never touched the inbox");
}
@Test
@@ -326,6 +333,26 @@ class FleetMcpTest {
assertEquals("orphan", drained.getFirst().content());
}
@Test
void replyWithBlankContentIsACleanToolErrorNotAnUncaughtException() {
// fleetd #302: MessageService.reply now REJECTS blank content by throwing. fleet_reply's
// handler is a bare BiFunction with no try/catch around it, so if this guard only checked
// `== null` (as it did), a whitespace-only reply would leave the handler as an uncaught
// IllegalArgumentException instead of a tool error the caller can read. Null and whitespace
// are the same caller mistake and must get the same answer — the sibling fleet_send guard
// has always used isBlank for exactly this reason.
for (String blank : new String[] {null, "", " ", "\n\t"}) {
McpSchema.CallToolResult res = assertDoesNotThrow(
() -> FleetMcp.reply(messages, "term_a", blank),
"blank content must be refused as a tool error, never thrown out of the handler");
assertEquals(Boolean.TRUE, res.isError(), "blank content is an error result");
assertTrue(textOf(res).contains("content is required"),
"the error names the missing argument: " + textOf(res));
}
assertEquals(0, messages.drainReplies("term_a").size(),
"a refused reply must not reach the inbox");
}
@Test
void bridgePollWithTargetDrainsReplies() {
// A reply with no open send queues it in the inbox.
@@ -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();
@@ -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;
}
}
@@ -940,6 +940,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");
@@ -957,6 +1005,76 @@ class MessageServiceTest {
assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
}
/**
* fleetd #307: a worker's {@code fleet_ask} can time out because the primary never answers —
* distinct from {@link #aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket}, where the
* primary DID answer and only its own bounded wait for the resumed turn expired.
* {@code ask()}'s timeout path deliberately forgets the task's {@code turnId} (so
* {@code hasAsyncQuestion} stops reporting the target BUSY — see
* {@code unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget} above), which used
* to also erase the one signal {@code askAnsweredAsyncTasks} needed to recognize the worker's
* eventual real {@code fleet_reply}. That reply then had nowhere to land but the inbox, and
* {@code fleet_poll{ticket}} stayed PENDING forever — later force-failed with the false reason
* "session released before it replied", even though the worker had, in fact, replied.
*/
@Test
void aReplyAfterAnAskTimeoutStillCompletesTheAsyncTicket() throws Exception {
String ticket = messages.sendAsync(T, "task that asks then finishes alone");
awaitWaiting();
injectDelivery();
assertEquals(MessageService.AskOutcome.TIMED_OUT,
messages.ask(T, "which config?", 200).outcome());
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
"only the question wait ended; the delegated turn may still finish");
// The worker keeps working past the timeout and only now calls fleet_reply — with no live
// rendezvous waiter open (ask()'s timeout already closed it) and no new send() having
// reopened one for this target.
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("PR opened: https://example/pulls/42", done.reply(),
"fleet_poll{ticket} must return the worker's real reply, not stay pending forever");
assertEquals("reply", done.replySource());
assertFalse(messages.hasStrandedReply(T),
"the reply completed its own ticket directly and never touched the inbox");
}
/**
* fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
* becomes false the instant it lapses — proven above), so a second, independent delegation can
* be dispatched to the same target and itself go on to ask-and-lapse before the first worker's
* real reply ever arrives. Two open tasks are then both eligible candidates on one target with
* no live waiter to disambiguate them. A reply arriving now must not guess which one it answers
* — guessing wrong would hand the lead a plausible-looking answer to a delegation the worker
* never touched, worse than a failure because the lead acts on it — so it must fall back to the
* inbox exactly as the zero-candidate case does.
*/
@Test
void twoAskTimedOutTicketsOnOneTargetFallBackToTheInboxRatherThanGuess() throws Exception {
String ticket1 = messages.sendAsync(T, "first task that asks");
awaitWaiting();
injectDelivery();
assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q1?", 200).outcome());
String ticket2 = messages.sendAsync(T, "second task that asks");
awaitWaiting();
injectDelivery();
assertEquals(MessageService.AskOutcome.TIMED_OUT, messages.ask(T, "Q2?", 200).outcome());
assertTrue(messages.reply(T, "which task does this answer?"));
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket1).phase(),
"an ambiguous reply must not guess ticket1");
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket2).phase(),
"an ambiguous reply must not guess ticket2");
assertTrue(messages.hasStrandedReply(T));
var drained = messages.drainReplies(T);
assertEquals(1, drained.size());
assertEquals("which task does this answer?", drained.get(0).content());
}
@Test
void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
String first = messages.sendAsync(T, "first task");
@@ -415,7 +415,11 @@ class FleetAppTest {
FakeHerdr herdr = new FakeHerdr().agentNameTakenTimes(99);
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
assertEquals(500, req(port, "POST", "/members").statusCode());
// fleetd #304: 502, not Javalin's default 500 — the herdr failure is named, and the body
// carries herdr's own message, matching what fleet_spawn reports for the same failure.
HttpResponse<String> res = req(port, "POST", "/members");
assertEquals(502, res.statusCode());
assertEquals("herdr_error", mapper.readTree(res.body()).get("error").asText());
assertTrue(herdr.called("tab.create"), "a tab was created before the failed start");
assertEquals("w9:t2", params(herdr, "tab.close").get("tab_id"), "orphaned tab must be closed");
}
@@ -733,7 +737,12 @@ class FleetAppTest {
int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
// A genuine teardown failure must surface, not be reported as a successful 204.
assertEquals(500, req(port, "DELETE", "/members/w9:pW").statusCode());
// fleetd #304: it surfaces as a named 502 rather than Javalin's default 500. The property
// this test guards is "not 204" and the herdr detail reaching the caller — a bare 500 gave
// the body "Server Error" and said nothing about herdr.
HttpResponse<String> res = req(port, "DELETE", "/members/w9:pW");
assertEquals(502, res.statusCode());
assertEquals("herdr_error", mapper.readTree(res.body()).get("error").asText());
assertFalse(herdr.called("tab.close"), "tab is not removed when the pane close failed");
}
@@ -746,4 +755,5 @@ class FleetAppTest {
assertEquals(204, req(port, "DELETE", "/members/w9:pW").statusCode());
assertTrue(herdr.called("tab.close"));
}
}
@@ -403,6 +403,52 @@ class GitWorktreesTest {
assertTrue(heads.isBlank(), "the branch leaked after a post-creation step threw:\n" + heads);
}
/**
* fleetd #309. The add runner is a narrow seam for the case where Git has created state but
* the caller then kills the process. The runner first performs the real add in this throwaway
* repo, then throws the same kind of exception that {@link GitWorktrees#exec} uses for a timeout.
* This proves the failure path cleans both real Git objects without waiting for a slow checkout.
*/
@Test
void addCleansUpWhenTheWorktreeAddRunnerFailsAfterCreatingState(@TempDir Path tmp) throws Exception {
Path repo = initRepo(tmp.resolve("repo"));
String branch = "cb-309-timeout";
WorktreeException timeout = new WorktreeException("command timed out: synthetic git worktree add");
AtomicReference<String> createdPath = new AtomicReference<>();
GitWorktrees worktrees = new GitWorktrees(tmp.resolve("wts").toString(), null, null, null, command -> {
createdPath.set(command[5]);
try {
git(repo, "worktree", "add", command[5], "-b", command[7], command[8]);
} catch (Exception e) {
throw new AssertionError("test setup could not create the worktree", e);
}
throw timeout;
});
WorktreeException thrown = assertThrows(WorktreeException.class,
() -> worktrees.add(repo.toString(), branch, "HEAD"));
assertSame(timeout, thrown, "cleanup must not replace the add failure");
assertNotNull(createdPath.get(), "the add runner must receive the worktree path");
assertFalse(Files.exists(Path.of(createdPath.get())),
"the worktree directory leaked after the add runner failed");
assertFalse(refExists(repo, "refs/heads/" + branch), "the branch leaked after the add runner failed");
}
/** An ordinary Git refusal must not delete the existing branch or log a cleanup warning. */
@Test
void addFailureBeforeCreatingAWorktreeIsQuiet(@TempDir Path tmp) throws Exception {
Path repo = initRepo(tmp.resolve("repo"));
String branch = "already-exists";
git(repo, "branch", branch);
assertThrows(WorktreeException.class, () -> new GitWorktrees(tmp.resolve("wts").toString())
.add(repo.toString(), branch, "HEAD"));
assertTrue(refExists(repo, "refs/heads/" + branch), "the existing branch must remain");
assertTrue(capturedMessages().isEmpty(), "an ordinary Git refusal logged a warning: " + capturedMessages());
}
// ---- CB-189: broader remote-URL coverage — every remote, both fetch and push URLs, any
// non-SSH scheme. Reporting only, additive to the origin/https strip-and-refuse tests above. ----
@@ -26,7 +26,12 @@ import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.LongSupplier;
import static org.junit.jupiter.api.Assertions.*;
@@ -81,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;
@@ -122,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;
}
@@ -178,14 +223,19 @@ class SessionManagerTest {
}
private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
boolean clearAfterTurn) {
boolean clearAfterTurn) {
return sessionManager(herdr, clock, contextCap, clearAfterTurn, null);
}
private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
boolean clearAfterTurn, java.util.function.Consumer<MemberSession> hook) {
FleetConfig.Profile cfg = new FleetConfig.Profile(
"ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
"worker: {profile} #{n}", null, null, null);
ClaudeCodeLauncher workers = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn);
return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn, hook);
}
@Test
@@ -670,6 +720,24 @@ class SessionManagerTest {
"BUSY session remains");
}
@Test
void reapIdleDoesNotReleaseSessionDeliveredAfterItsEligibilityCheck() {
long[] clock = {0};
FakeHerdr herdr = new FakeHerdr();
SessionManager[] manager = new SessionManager[1];
SessionManager sessions = sessionManager(herdr, () -> clock[0], 0, false,
session -> manager[0].onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId())));
manager[0] = sessions;
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
sessions.asPresence().markPresent(session.terminalId());
clock[0] = 11;
assertEquals(0, sessions.reapIdle(10), "delivery replaces the idle snapshot before release");
assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
"a just-delivered session stays registered and busy");
assertFalse(herdr.called("pane.close"), "the busy session pane is not stopped");
}
@Test
void doneSessionPastIdleTtlIsReaped() {
long[] clock = {0};
@@ -824,6 +892,186 @@ class SessionManagerTest {
.count();
}
// --- fleetd #308: a spawn accepted while the shutdown drain is running must not orphan ---
@Test
void acquireRefusesANewSpawnOnceDrainAllHasStarted() {
FakeHerdr herdr = new FakeHerdr();
SessionManager sessions = sessionManager(herdr);
sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(50)); // empty roster — returns immediately,
// but the shutdown guard it flips must stay tripped for the life of the process.
ShuttingDownException e = assertThrows(ShuttingDownException.class,
() -> sessions.acquire("ltms-local", null, "/caller", "term_primary"),
"a spawn requested after the drain has begun must be refused loudly (invariant 3), "
+ "not silently registered into a registry the drain will never revisit");
assertNotNull(e.getMessage());
assertFalse(e.getMessage().isBlank(), "the refusal must say why, not just that it failed");
assertTrue(sessions.roster().isEmpty(), "the refused spawn must never reach the registry");
}
/**
* fleetd #308: the guard above closes most of the shutdown-race window, but it cannot close
* all of it — a caller that already passed the {@code draining} check before {@code drainAll}
* flips it can still be mid-{@code launcher.spawn()} (a real herdr round trip, not
* instantaneous) when {@code drainAll} takes its registry snapshot. This test forces exactly
* that interleaving with a launcher double that blocks the second {@code spawn()} call and the
* first {@code stop()} call until released, then proves the post-loop sweep in {@code
* drainAll} still finds and tears down the straggler that lands in the registry afterward.
*/
@Test
void drainAllSweepsAStragglerThatRegisteredAfterTheInitialSnapshot() throws Exception {
FakeHerdr herdr = new FakeHerdr();
FleetConfig.Profile cfg = new FleetConfig.Profile(
"ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
"worker: {profile} #{n}", null, null, null);
ClaudeCodeLauncher delegate = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
RaceLauncher race = new RaceLauncher(delegate);
SessionManager sessions = new SessionManager(race);
// Registered normally, before the drain starts — the first spawn call, never blocked.
MemberSession ready = sessions.acquire("ltms-local", "/ready", "/caller", "ownerR");
ExecutorService exec = Executors.newFixedThreadPool(2);
try {
// The straggler's acquire() reads `draining == false` (checked before this call ever
// touches the launcher) and then blocks inside its own spawn() — the second spawn call.
Future<MemberSession> straggler = exec.submit(() ->
sessions.acquire("ltms-local", "/late", "/caller", "ownerLate"));
assertTrue(race.enteredSecondSpawn.await(5, TimeUnit.SECONDS),
"the straggler must have passed the shutdown guard and reached spawn() before "
+ "drainAll ever runs");
assertEquals(1, sessions.roster().size(),
"the straggler is still inside spawn() — not registered yet");
// drainAll flips `draining`, snapshots the registry (only `ready` is in it), and starts
// releasing that snapshot — its first release() call stops `ready`'s pane, which this
// launcher double blocks on so the interleaving below is deterministic, not a timing bet.
Future<?> drain = exec.submit(() -> sessions.drainAll(TimeUnit.SECONDS.toNanos(5)));
assertTrue(race.enteredFirstStop.await(5, TimeUnit.SECONDS),
"drainAll must be stopping the ready session's pane — proof its initial "
+ "registry snapshot has already been taken");
// Only now does the straggler's spawn complete and register — strictly after the
// snapshot drainAll's main pass is working from.
race.releaseSecondSpawn.countDown();
MemberSession registered = straggler.get(5, TimeUnit.SECONDS);
// Let drainAll finish releasing `ready`; it then re-checks the registry and must find
// (and drain) the straggler that just landed in it.
race.releaseFirstStop.countDown();
drain.get(5, TimeUnit.SECONDS);
assertTrue(sessions.roster().isEmpty(),
"the post-loop sweep must drain the straggler too, not just the initial snapshot");
assertNotNull(registered.paneId());
long paneCloseCalls = herdr.calls.stream().filter(c -> "pane.close".equals(c.method())).count();
assertEquals(2, paneCloseCalls,
"both ready's pane AND the straggler's pane must actually be stopped — a pane "
+ "left running is exactly the orphan this ticket is about");
} finally {
exec.shutdownNow();
}
}
/**
* Delegates every call while blocking the SECOND {@code spawn()} call and the FIRST
* {@code stop()} call until the test releases them — used to force the fleetd #308 race
* deterministically instead of betting on real thread-scheduling timing.
*/
private static final class RaceLauncher implements PeerLauncher {
private final PeerLauncher delegate;
private final AtomicInteger spawnCalls = new AtomicInteger();
private final AtomicInteger stopCalls = new AtomicInteger();
final CountDownLatch enteredSecondSpawn = new CountDownLatch(1);
final CountDownLatch releaseSecondSpawn = new CountDownLatch(1);
final CountDownLatch enteredFirstStop = new CountDownLatch(1);
final CountDownLatch releaseFirstStop = new CountDownLatch(1);
RaceLauncher(PeerLauncher delegate) {
this.delegate = delegate;
}
private static void awaitOrFail(CountDownLatch latch) {
try {
if (!latch.await(5, TimeUnit.SECONDS)) {
throw new AssertionError("RaceLauncher latch timed out");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError("RaceLauncher latch interrupted", e);
}
}
@Override
public Set<Capability> capabilities() {
return delegate.capabilities();
}
@Override
public Set<Capability> capabilitiesFor(String profileName) {
return delegate.capabilitiesFor(profileName);
}
@Override
public PeerHandle spawn(SpawnRequest req) {
if (spawnCalls.incrementAndGet() == 2) {
enteredSecondSpawn.countDown();
awaitOrFail(releaseSecondSpawn);
}
return delegate.spawn(req);
}
@Override
public Set<String> profiles() {
return delegate.profiles();
}
@Override
public String defaultProfile() {
return delegate.defaultProfile();
}
@Override
public String effectiveCwd(SpawnRequest req) {
return delegate.effectiveCwd(req);
}
@Override
public List<String> parityOverlay(String profileName) {
return delegate.parityOverlay(profileName);
}
@Override
public List<?> list() {
return delegate.list();
}
@Override
public int reapOrphanWorkers() {
return delegate.reapOrphanWorkers();
}
@Override
public void stop(String id) {
if (stopCalls.incrementAndGet() == 1) {
enteredFirstStop.countDown();
awaitOrFail(releaseFirstStop);
}
delegate.stop(id);
}
@Override
public boolean clearContext(String id) {
return delegate.clearContext(id);
}
}
private static List<String> promptTexts(FakeHerdr herdr) {
return herdr.calls.stream()
.filter(c -> "agent.prompt".equals(c.method()))
@@ -999,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