Compare commits

..

1 Commits

Author SHA1 Message Date
Dai Ha 6ed70700a0 #282: don't let a chained fleet_ask kill its own async ticket
CI / contract (pull_request) Successful in 1m6s
CI / build (pull_request) Successful in 1m38s
answer() opened a fresh forward waiter but, unlike send(), never
registered it in asyncTasksByWaiter. So when a worker chained a
second fleet_ask inside the same resumed turn (before calling
fleet_reply), markAsyncQuestion had no Task to re-associate, and
answer() then completed the async ticket's future with the second
QUESTION as if it were a terminal reply — fleet_poll reported FAILED
while the worker was still alive and mid-conversation.

Fix: register answer()'s waiter in asyncTasksByWaiter (mirroring
send()) so a chained ask can re-arm the ticket under its new turnId,
and guard answer()'s finishAsyncTask call the same way sendAsync's
own lambda already does (skip on Outcome.QUESTION). Also drop the
stale asyncTasksByTurn entry left behind when markAsyncQuestion
re-arms a task under a new turnId, a leak the fix makes reachable
for the first time.

Reachability confirmed by driving the exact sequence through the
public API (sendAsync -> ask -> answer -> ask again) in a new test;
reverting the production change makes it fail with
"expected: <ASKING> but was: <FAILED>", confirming it catches the
regression.
2026-09-04 10:52:54 +07:00
4 changed files with 99 additions and 237 deletions
@@ -18,9 +18,7 @@ import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.attribute.GroupPrincipal;
import java.nio.file.attribute.PosixFileAttributeView;
import java.nio.file.attribute.PosixFilePermissions;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.List;
@@ -542,65 +540,16 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
* itself is still lost, because there is no OS-level compare-and-swap on a plain file, only this
* cooperative narrowing of the gap.
*
* <p><b>fleetd #285: refuses under {@code memberHerdrSocket} rather than writing somewhere the
* member cannot read.</b> Under {@code memberHerdrSocket:} the member pane runs as a
* <em>different OS user with its own {@code $HOME}</em> — the same reason {@link
* #writeCharterFile} routes the role/reply charter under {@code worktreeRoot} instead of
* {@code java.io.tmpdir} and refuses the spawn when it cannot. This method has no equivalent
* relocation available: unlike the charter (fleetd's own content, free to place anywhere and
* hand to the peer via an argv flag), {@code .claude.json} is a file Claude Code looks up for
* ITSELF at a fixed location — {@code CLAUDE_CONFIG_DIR/.claude.json}, or else the member OS
* user's own {@code ~/.claude.json}, a path fleetd has no channel to learn. So when {@code
* configDir} is unset, there is no member-readable target to seed at all — writing the
* unqualified default would land in <em>fleetd's own</em> {@code ~/.claude.json} instead, the
* exact defect this fix closes, not a workable fallback. And even with {@code configDir} set,
* the file this method itself just wrote is {@code 0600} (owner-only — see {@link
* #copyPosixPermissionsIfPresent}), unreadable by a different-uid member unless shared with
* {@code worktreeGroup}, the same group {@link EnvAllowListScrub#shareWithGroup} already uses
* for the ZDOTDIR scrub (fleetd #213) and the charter file (fleetd #219/#222). So under {@code
* memberHerdrSocket} this method requires BOTH {@code configDir} and {@code worktreeGroup}
* before it ever touches a file, and refuses the spawn — naming exactly which one is missing —
* rather than silently corrupt fleetd's own home or hand the member an unreadable path. This
* mirrors {@link #writeCharterFile}'s "refuse, don't degrade" decision: a member spawned without
* a readable trust seed is not degraded, it sits on the interactive dialog forever and never
* calls {@code fleet_reply} — exactly the failure fleetd #149 exists to prevent, so trading it
* for "spawn something" is not worth it. When {@code configDir} and {@code worktreeGroup} are
* both present, the write proceeds exactly as below and the resulting file is additionally
* chgrp'd/chmod'd group-readable ({@code rw-r-----}) via {@link
* #shareTrustJsonWithGroup(Path, String)} so the member's OS user can actually open it — the
* directory itself (unlike {@code worktreeRoot} or the charter's per-spawn directory) is not
* fleetd-managed, so its own traversal permissions remain the operator's setup, same as they
* already must be for the member to read anything else fleetd points {@code CLAUDE_CONFIG_DIR}
* at. With {@code memberHerdrSocket} ABSENT (today's only live mode) every branch below is
* byte-identical to before this fix.
*
* @param configDir the profile's {@code CLAUDE_CONFIG_DIR} ({@code cfg.configDir()}), or
* {@code null}/blank to target the default {@code ~/.claude.json} — refused
* outright when {@code memberHerdrSocket} is configured, see above
* {@code null}/blank to target the default {@code ~/.claude.json}
* @param cwd the spawn's resolved working directory — the exact key Claude Code will look
* up for itself once it starts there
* @throws IllegalStateException when {@code memberHerdrSocket} is configured but {@code
* configDir} and/or {@code worktreeGroup} is not — the same
* refusal shape as {@link #writeCharterFile}
*/
private void seedTrustDialog(String configDir, String cwd) {
private static void seedTrustDialog(String configDir, String cwd) {
if (!isProvisionedWorktree(cwd)) {
return;
}
boolean unsetConfigDir = configDir == null || configDir.isBlank();
boolean memberHerdrSocket = memberHerdrSocketConfigured();
String group = memberHerdrSocket ? memberGroup() : null;
if (memberHerdrSocket && (unsetConfigDir || group == null)) {
throw new IllegalStateException("memberHerdrSocket is configured, so the workspace-trust "
+ "seed (.claude.json, which gates Claude Code's interactive trust dialog) must be "
+ "placed where the member's OS user can read it — configDir, shared via "
+ "worktreeGroup — but " + (unsetConfigDir ? "configDir" : "worktreeGroup")
+ " is not configured. Refusing to spawn rather than write fleetd's own default "
+ "'~/.claude.json' or hand the member a config file it cannot read: that member "
+ "would sit on the interactive trust dialog forever and never reach an "
+ "injectable state. Configure configDir on this profile and worktreeGroup on the "
+ "fleet to enable claude-code member spawns under memberHerdrSocket.");
}
Path target = unsetConfigDir
? Path.of(System.getProperty("user.home"), ".claude.json")
: Path.of(configDir, ".claude.json");
@@ -610,20 +559,18 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
// profile that simply forgot to set configDir gets no signal at all short of the
// operator noticing their own file changing. Say so loudly, every time it is about to
// happen, rather than only once ever: each occurrence is a live write to a real
// person's home config and deserves its own log line. (Reached only when
// memberHerdrSocket is absent — the block above already refused otherwise.)
// person's home config and deserves its own log line.
log.warn("seedTrustDialog: profile has no configDir set, so the workspace-trust seed "
+ "for cwd '{}' is about to write the operator's own default '{}' — set "
+ "configDir on this profile to target a per-member config file instead",
cwd, target);
}
synchronized (TRUST_JSON_LOCK) {
boolean written = false;
try {
if (target.getParent() != null) {
Files.createDirectories(target.getParent());
}
for (int attempt = 1; attempt <= MAX_TRUST_JSON_CAS_ATTEMPTS && !written; attempt++) {
for (int attempt = 1; attempt <= MAX_TRUST_JSON_CAS_ATTEMPTS; attempt++) {
byte[] before = Files.isRegularFile(target) ? Files.readAllBytes(target) : null;
ObjectNode root = parseTrustJsonOrEmpty(before);
JsonNode projectsNode = root.get("projects");
@@ -664,67 +611,23 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
continue;
}
writeAtomically(target, newContent);
written = true;
}
if (!written) {
// fleetd #247: deliberately do NOT write here. A member that starts without the
// seed still starts — it may hit the trust dialog fleetd #149 describes and fail
// to reach an injectable state, but that failure is visible (herdr reports it,
// the spawn-readiness gate times out) and recoverable (retry the spawn). Writing
// our stale copy over whatever the other writer left would be silent and, if that
// other writer is the operator's own live session, could destroy real
// configuration — fail toward the recoverable outcome, not the silent one.
log.warn("seedTrustDialog: gave up seeding workspace-trust for cwd '{}' into '{}' "
+ "after {} attempts — another writer (most plausibly the operator's own "
+ "live Claude Code sharing this file) kept changing it faster than we "
+ "could re-read it, so nothing was written; the member may show the "
+ "trust dialog instead", cwd, target, MAX_TRUST_JSON_CAS_ATTEMPTS);
return;
}
// fleetd #247: deliberately do NOT write here. A member that starts without the
// seed still starts — it may hit the trust dialog fleetd #149 describes and fail to
// reach an injectable state, but that failure is visible (herdr reports it, the
// spawn-readiness gate times out) and recoverable (retry the spawn). Writing our
// stale copy over whatever the other writer left would be silent and, if that other
// writer is the operator's own live session, could destroy real configuration —
// fail toward the recoverable outcome, not the silent one.
log.warn("seedTrustDialog: gave up seeding workspace-trust for cwd '{}' into '{}' "
+ "after {} attempts — another writer (most plausibly the operator's own "
+ "live Claude Code sharing this file) kept changing it faster than we could "
+ "re-read it, so nothing was written; the member may show the trust dialog "
+ "instead", cwd, target, MAX_TRUST_JSON_CAS_ATTEMPTS);
} catch (Exception e) {
log.debug("cannot seed workspace-trust entry for cwd '{}' into '{}'", cwd, target, e);
return;
}
// fleetd #285: the write above lands as fleetd's own OS user; under memberHerdrSocket
// that is NOT the member's OS user, so without this the member still cannot read the
// file it exists to seed — a silent readiness timeout with the write looking "done".
// Deliberately OUTSIDE the swallow-all catch above: a group that fails to resolve here
// means the seed is unreadable despite a successful write, which must fail as loudly as
// writeCharterFile's own EnvAllowListScrub.shareWithGroup call already does.
if (written && memberHerdrSocket) {
shareTrustJsonWithGroup(target, group);
}
}
}
/**
* fleetd #285: chgrp + chmod {@code target} — a {@code .claude.json} this method just wrote as
* fleetd's own OS user — group-readable ({@code rw-r-----}) for {@code group}, so a member OS
* user under {@code memberHerdrSocket} can open it. Mirrors {@link
* EnvAllowListScrub#shareWithGroup}'s per-file mode exactly (read-only for the group — a member
* never needs to write this file itself), but touches only the FILE, never its parent
* directory: unlike {@code worktreeRoot} or the charter's per-spawn directory, {@code
* configDir} is not a directory fleetd creates or owns, so its traversal permissions are left to
* the operator's own setup, same as they already must be for the member to read anything else
* fleetd points {@code CLAUDE_CONFIG_DIR} at.
*
* @throws UncheckedIOException when {@code group} does not resolve on this host, or a
* group-ownership/permission call is refused — the file was written
* but is not usable by the member, and that must be loud
*/
private static void shareTrustJsonWithGroup(Path target, String group) {
try {
GroupPrincipal principal = target.getFileSystem().getUserPrincipalLookupService()
.lookupPrincipalByGroupName(group);
PosixFileAttributeView view = Files.getFileAttributeView(target, PosixFileAttributeView.class);
if (view == null) {
throw new IOException("POSIX file attributes are not supported for " + target);
}
view.setGroup(principal);
Files.setPosixFilePermissions(target, PosixFilePermissions.fromString("rw-r-----"));
} catch (IOException e) {
throw new UncheckedIOException("cannot share workspace-trust file " + target + " with "
+ "group '" + group + "' — the group must exist, and the fleetd operator ("
+ System.getProperty("user.name") + ") must be a member of it", e);
}
}
@@ -927,7 +927,16 @@ public final class MessageService {
}
try {
CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(workerSession);
// #282: mirror send()'s registration (:802) so a SECOND fleet_ask inside this same
// resumed turn can re-associate the async ticket with its new turnId via
// markAsyncQuestion — without this, that second ask has no Task to attach to, and
// markAsyncQuestion silently returns null.
Task task = asyncTasksByTurn.get(turnId);
if (task != null) {
asyncTasksByWaiter.put(reply, task);
}
if (!rendezvous.answerAsk(turnId, content)) {
asyncTasksByWaiter.remove(reply);
rendezvous.close(workerSession, reply);
return new Reply(Outcome.STALE_TURN, null); // lapsed between the lookup and the unblock
}
@@ -935,7 +944,13 @@ public final class MessageService {
try {
Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
Reply result = new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
finishAsyncTask(turnId, result);
// #282: this waiter can resolve with a FRESH question rather than a terminal reply —
// the worker chained a second fleet_ask before replying. Mirror sendAsync's own guard
// (:1000) and leave the ticket open (markAsyncQuestion above already re-armed it under
// the new turnId) instead of completing it here with a QUESTION "reply".
if (result.outcome() != Outcome.QUESTION) {
finishAsyncTask(turnId, result);
}
return result;
} catch (TimeoutException e) {
// The worker resumed but hasn't replied yet — no completion fallback arms an answered
@@ -948,6 +963,7 @@ public final class MessageService {
Thread.currentThread().interrupt();
throw new IllegalStateException("interrupted awaiting reply from " + workerSession, e);
} finally {
asyncTasksByWaiter.remove(reply);
rendezvous.close(workerSession, reply);
}
} finally {
@@ -1106,9 +1122,16 @@ public final class MessageService {
private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
if (task != null) {
String previousTurnId = task.turnId;
task.question = new Reply(Outcome.QUESTION, text, turnId);
task.turnId = turnId;
asyncTasksByTurn.put(turnId, task);
// #282: a second fleet_ask in the same resumed turn re-arms an already-answered task
// (answer() re-registers it in asyncTasksByWaiter) under a FRESH turnId — drop the old
// key so asyncTasksByTurn does not keep growing by one stale entry per chained ask.
if (previousTurnId != null && !previousTurnId.equals(turnId)) {
asyncTasksByTurn.remove(previousTurnId, task);
}
}
return task;
}
@@ -2742,126 +2742,4 @@ class ClaudeCodeLauncherTest {
"a WARN naming the cwd must fire when the profile sets no configDir: "
+ appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList());
}
// --- fleetd #285: seedTrustDialog under memberHerdrSocket --------------------------------------
//
// seedTrustDialog gated only on isProvisionedWorktree(cwd) and, being static, could not see
// memberHerdrSocketConfigured() at all — unlike its sibling writeCharterFile one method below,
// which already refuses the spawn when it cannot place the charter where a different-uid member
// can read it. Under memberHerdrSocket + configDir unset, seedTrustDialog wrote fleetd's OWN
// ~/.claude.json (the operator's real file) while believing it was seeding the member's. The fix
// makes the method an instance method so it can see memberHerdrSocketConfigured(), and applies
// the same "refuse, don't silently write somewhere wrong" rule writeCharterFile already uses.
@Test
void seedTrustDialogUnderMemberHerdrSocketSharesTheFileWithTheConfiguredGroup(
@TempDir Path configDir, @TempDir Path worktree, @TempDir Path worktreeRoot) throws Exception {
markAsProvisionedWorktree(worktree);
String group = currentUserGroup();
FakeHerdr herdr = new FakeHerdr();
// trustProfile always sets a bridge mcpUrl, so a reply charter is generated too, which
// means writeCharterFile ALSO runs under memberHerdrSocket and needs its own worktreeRoot —
// pass one so this test isolates the trust-seed behaviour instead of tripping that refusal.
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
serviceWithConfig(herdr, cfg, () -> configWithMemberHerdrSocket(worktreeRoot.toString(), group)).spawn();
Path claudeJson = configDir.resolve(".claude.json");
assertTrue(Files.exists(claudeJson), "still seeded into <configDir>/.claude.json under memberHerdrSocket");
JsonNode project = new ObjectMapper().readTree(claudeJson.toFile())
.path("projects").path(worktree.toString());
assertTrue(project.path("hasTrustDialogAccepted").asBoolean(false));
assertEquals("rw-r-----", PosixFilePermissions.toString(Files.getPosixFilePermissions(claudeJson)),
"under memberHerdrSocket the file must be shared group-readable, mirroring the "
+ "charter file's own per-file mode (fleetd #222) — a 0600 file (Claude "
+ "Code's own default) is unreadable by the member's different OS user");
String actualGroup = Files.getFileAttributeView(claudeJson, PosixFileAttributeView.class)
.readAttributes().group().getName();
assertEquals(group, actualGroup, "the file must be chgrp'd to the configured worktreeGroup");
}
/**
* The bug itself: {@code memberHerdrSocket} configured, {@code configDir} unset. Before the fix
* this wrote fleetd's own default {@code ~/.claude.json} (here redirected to {@code fakeHome} so
* a reintroduced bug still cannot touch the real operator file); after the fix it must refuse the
* spawn instead, naming {@code configDir} as the missing key, before the member is ever started.
*/
@Test
void seedTrustDialogUnderMemberHerdrSocketRefusesWhenConfigDirUnset(
@TempDir Path fakeHome, @TempDir Path worktree) throws Exception {
markAsProvisionedWorktree(worktree);
String originalHome = System.getProperty("user.home");
System.setProperty("user.home", fakeHome.toString());
try {
FakeHerdr herdr = new FakeHerdr();
FleetConfig.Profile cfg = trustProfile(null, worktree.toString());
ClaudeCodeLauncher launcher = serviceWithConfig(herdr, cfg,
() -> configWithMemberHerdrSocket(null, "some-group"));
IllegalStateException ex = assertThrows(IllegalStateException.class, launcher::spawn,
"memberHerdrSocket + no configDir must refuse the spawn, not write fleetd's own "
+ "default ~/.claude.json");
assertTrue(ex.getMessage().contains("configDir"),
"the refusal must name the missing config key — got: " + ex.getMessage());
assertFalse(Files.exists(fakeHome.resolve(".claude.json")),
"nothing may be written to fleetd's own default home — this is the exact fleetd "
+ "#285 defect: writing the operator's own home instead of the member's");
assertFalse(herdr.called("agent.start"),
"the spawn must be refused BEFORE the member is ever started — got calls: " + herdr.calls);
} finally {
System.setProperty("user.home", originalHome);
}
}
/**
* {@code configDir} alone is not enough — without {@code worktreeGroup} the file fleetd writes
* stays {@code 0600} and the member's different OS user still cannot read it, so this must also
* refuse, naming {@code worktreeGroup} this time.
*/
@Test
void seedTrustDialogUnderMemberHerdrSocketRefusesWhenWorktreeGroupUnset(
@TempDir Path configDir, @TempDir Path worktree) throws Exception {
markAsProvisionedWorktree(worktree);
FakeHerdr herdr = new FakeHerdr();
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
ClaudeCodeLauncher launcher = serviceWithConfig(herdr, cfg,
() -> configWithMemberHerdrSocket(null, null));
IllegalStateException ex = assertThrows(IllegalStateException.class, launcher::spawn,
"memberHerdrSocket + no worktreeGroup must refuse the spawn, not write an unreadable file");
assertTrue(ex.getMessage().contains("worktreeGroup"),
"configDir alone is not enough — got: " + ex.getMessage());
assertFalse(Files.exists(configDir.resolve(".claude.json")),
"nothing may be written when the file cannot be shared with the member's group");
assertFalse(herdr.called("agent.start"),
"the spawn must be refused BEFORE the member is ever started");
}
/**
* Regression proof: with {@code memberHerdrSocket} ABSENT — even given a LIVE, non-null {@code
* config} supplier (not merely {@code config == null}, which every other seedTrustDialog test in
* this file already exercises) — the write must stay byte-identical to before this fix: existing
* {@code 0600} permissions preserved, no chgrp/chmod attempted. This is the proof the ticket asks
* for: the path the whole live fleet uses today is unchanged by this fix.
*/
@Test
void seedTrustDialogPreservesExisting0600PermissionsWhenMemberHerdrSocketAbsentEvenWithALiveConfigSupplier(
@TempDir Path configDir, @TempDir Path worktree) throws Exception {
markAsProvisionedWorktree(worktree);
Path claudeJson = configDir.resolve(".claude.json");
Files.writeString(claudeJson, "{}");
assumeTrue(Files.getFileAttributeView(claudeJson, PosixFileAttributeView.class) != null,
"no POSIX permissions on this filesystem — skipping rather than failing");
Files.setPosixFilePermissions(claudeJson, PosixFilePermissions.fromString("rw-------"));
FakeHerdr herdr = new FakeHerdr();
FleetConfig.Profile cfg = trustProfile(configDir.toString(), worktree.toString());
FleetConfig config = new FleetConfig(null, null, null, Map.of(), null, null, null, null, null,
null, null, null, null, null, null, null, null, null, null, null, null, null).withDefaults();
serviceWithConfig(herdr, cfg, () -> config).spawn();
assertEquals("rw-------", PosixFilePermissions.toString(Files.getPosixFilePermissions(claudeJson)),
"with memberHerdrSocket absent — even given a live config supplier — the seed must "
+ "stay byte-identical to before this fix: no chgrp/chmod attempted");
}
}
@@ -975,6 +975,64 @@ class MessageServiceTest {
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
}
/**
* fleetd #282: a worker that chains a SECOND {@code fleet_ask} inside the same resumed turn —
* before it ever calls {@code fleet_reply} — used to kill its own async ticket. {@code answer()}
* opens a fresh forward waiter but (unlike {@code send()}) never registered it in
* {@code asyncTasksByWaiter}, so the second ask's {@code markAsyncQuestion} found no {@code Task}
* to re-associate. That waiter still resolved with the second {@code QUESTION} once the worker
* asked again, and {@code answer()} completed the ticket's future with that QUESTION "reply"
* unconditionally — so {@code fleet_poll} reported FAILED while the worker was still alive and
* the primary was mid-conversation with it.
*
* <p>Driven entirely through {@code MessageService}'s public API (sendAsync/ask/answer/poll) —
* never by reaching into {@link Rendezvous} or the task maps directly, so this test cannot pass
* for a reason unrelated to the real bug.
*/
@Test
void secondFleetAskInTheSameResumedTurnDoesNotKillTheAsyncTicket() throws Exception {
String ticket = messages.sendAsync(T, "task that asks twice");
awaitWaiting();
// The worker's first fleet_ask.
CompletableFuture<MessageService.AskResult> ask1 =
CompletableFuture.supplyAsync(() -> messages.ask(T, "Q1", 5000));
MessageService.TaskView asking1 = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
assertEquals("Q1", asking1.reply());
// The primary answers it — answer() resumes the turn and blocks for what comes next.
CompletableFuture<MessageService.Reply> answer1 = CompletableFuture.supplyAsync(
() -> messages.answer(asking1.turnId(), "a1", 5000));
assertEquals("a1", ask1.get(5, TimeUnit.SECONDS).answer());
// Still in the SAME resumed turn — before replying — the worker asks again.
CompletableFuture<MessageService.AskResult> ask2 =
CompletableFuture.supplyAsync(() -> messages.ask(T, "Q2", 5000));
// answer1's own call unblocks with the second QUESTION (documented QUESTION-chaining
// behaviour — see FleetMcp.answer's javadoc: "Answer it by calling fleet_send again with
// turnId=..."). The bug: this used to also kill the async ticket in the process.
MessageService.Reply firstAnswerResult = answer1.get(5, TimeUnit.SECONDS);
assertEquals(MessageService.Outcome.QUESTION, firstAnswerResult.outcome());
String turnId2 = firstAnswerResult.turnId();
MessageService.TaskView asking2 = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
assertEquals("Q2", asking2.reply(),
"the ticket must surface the SECOND question, not be dead/FAILED");
assertEquals(turnId2, asking2.turnId());
// The primary answers the second question; the worker finally sends its real fleet_reply.
CompletableFuture<MessageService.Reply> answer2 = CompletableFuture.supplyAsync(
() -> messages.answer(turnId2, "a2", 5000));
assertEquals("a2", ask2.get(5, TimeUnit.SECONDS).answer());
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
assertEquals(MessageService.Outcome.REPLIED, answer2.get(5, TimeUnit.SECONDS).outcome());
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("done", done.reply());
}
// --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
@Test