#185 stage 2: stop the credential-gap detector reporting on the wrong environment

When memberHerdrSocket: is configured, member panes run under a different OS
user than fleetd's own process, so hostEnvNames (fleetd's own environment)
no longer describes what a member pane inherits. logCredentialGap now checks
for that config key and, when set, logs a single WARN saying the gap is
UNKNOWN (not clean) and names the key, instead of printing the "inherits
them UNBLOCKED" / "the scrub blanks them" conclusions as fact. Behaviour is
byte-identical when memberHerdrSocket is absent (the default and only mode
this host runs).

REPORT-cb137.md

@@ -1,166 +0,0 @@
-# CB-137 / fleetd issue #137 — report
-
-## Real root cause (not the hypothesis in the ticket)
-
-I read `MessageService.java` and `Rendezvous.java` before changing anything. The mechanism is real,
-but the exact place it happens is `MessageService.answer()`, not "the reply goes to the inbox on
-purpose" in general.
-
-1. A lead delegates with `fleet_send{wait:false}` → `sendAsync()` creates a `Task` and runs `send()`
-   on a background virtual thread with a 30-minute internal budget (`ASYNC_TIMEOUT_MS`).
-2. The worker calls `fleet_ask`. That resolves the open rendezvous waiter with `Kind.QUESTION`, so
-   `send()` returns immediately and the `Task` is left open (its `future` stays unresolved — see the
-   comment in `sendAsync`'s lambda: "Keep the accepted owner until answer() finishes it").
-3. The lead answers with `fleet_send{turnId, content}`. This calls `FleetMcp.answer()` →
-   `MessageService.answer(turnId, content, timeout)`. The `timeout` here is **not** the generous
-   30-minute async budget — it is the MCP tool's own bounded wait: `DEFAULT_TIMEOUT_MS = 25_000`,
-   clamped to at most `MAX_TIMEOUT_MS = 120_000` (`FleetMcp.java:71-72,495,512`). This is the same
-   ~60–120s window every blocking `fleet_send` call is capped at (documented elsewhere as "the
-   caller's own MCP client call timeout").
-4. `answer()` opens a **fresh** rendezvous waiter for the worker session and blocks on it for at most
-   that window. If the worker's resumed turn takes longer than that to actually finish (very
-   plausible — the resumed turn can mean more edits, a build, a commit, a push, opening a PR), the
-   wait times out. On timeout, `answer()`'s `finally` block unconditionally calls
-   `rendezvous.close(workerSession, reply)`, **removing the waiter from the map**, and returns
-   `Outcome.TIMED_OUT_WORKING` to the lead.
-5. The worker keeps working, unaware anything happened, and eventually calls `fleet_reply`. That
-   reaches `MessageService.reply(session, content)`, which tries `rendezvous.resolve(session,
-   content)` — but the waiter was already closed in step 4, so `resolve` returns `false`. `reply()`
-   then falls back to `inbox.publish(...)` and marks `strandedReplies.put(session, true)`
-   (CB-640 bookkeeping) — the reply is safely held, but **the async `Task`'s `future` is never
-   completed**.
-6. `fleet_poll{ticket}` keeps returning `PENDING` forever (the `Task` never resolves) — until the
-   lead eventually calls `fleet_stop`. That fires `sessions.onRelease` → `messages.abandon(target,
-   reason)` (`Fleetd.java:481-497`), where `reason` is built with the exact text from the bug report
-   ("the worker session was released before it replied; worktree=... branch=... snapshot=...",
-   `Fleetd.java:484-487`). `abandon()`'s loop finds the still-open `Task` (`question == null`, future
-   not done) and completes it as `WORKER_FAILED` with that misleading reason — even though the
-   worker's real reply is sitting, intact, in the inbox the whole time.
-
-So: the reported behaviour is correct, and the specific trigger is `answer()`'s own bounded wait
-being shorter than the worker's real resumed-turn time — not anything to do with the ~55s
-`fleet_ask` window itself (that part, issue #61, is untouched).
-
-## Fix
-
-Two changes in `fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java`, both scoped to the
-ticket/reply routing and the terminal-state text — `fleet_ask`'s own window and mechanics are
-untouched.
-
-**1. `reply()` — priority 1 (the ticket resolves with the real reply).**
-Before falling back to the inbox, `reply()` now looks for an async `Task` that is specifically in the
-"already answered but not yet resolved" state (`question == null`, `turnId != null` — set once
-`answer()` has cleared the question but before anything completed the future, `!future.isDone()`).
-If one exists for this `target`, the worker's reply completes that `Task`'s future directly as
-`Outcome.REPLIED` with the real content, and the reply never touches the inbox at all. A task that
-was never asked has `turnId == null` and can never match, so ordinary (no-`fleet_ask`) delegations
-are unaffected — they already resolve through the pre-existing rendezvous fast path.
-
-I chose this over leaving `answer()`'s own timeout behaviour untouched and instead keeping its
-rendezvous waiter open in the background: that alternative works but reopens the "at most one
-waiter per session" invariant (`Rendezvous.open` throws on a double-open) to a new class of races
-with a fresh send arriving mid-window. The `send()` path already guards against sending into an
-answered-but-still-resolving worker via `hasAsyncQuestion(target)` (checks `asyncTasksByTurn`,
-which still holds the task until it resolves), so routing through `reply()` gets the same protection
-without touching `answer()`'s waiter lifecycle at all — the smaller, safer diff.
-
-**2. `abandon()` — priority 3 (required independently, "even if you fix (1)").**
-Before marking any of a released target's still-open tasks `WORKER_FAILED`, `abandon()` now checks
-`hasStrandedReply(target)` (the existing CB-640 fact — true whenever the *last* `reply()` for this
-target fell through to the inbox). If true, it drains the inbox (`recoverStrandedReply`) and — if it
-actually finds a message — completes the task as `REPLIED` with that real content instead of writing
-the failure. This is deliberately a **separate** check from fix 1: fix 1 already prevents the
-inbox-stranding from happening in the exact scenario this ticket describes, so by the time
-`abandon()` runs the task is normally already resolved and `abandon()`'s `complete()` call is a
-harmless no-op. This second check exists so that if some *other* future path ever strands a reply
-in the inbox without resolving its ticket, `abandon()` still refuses to report a false failure —
-"if a reply reached any sink for that turn, the terminal state is done," per the ticket. I verified
-both are required by disabling each independently and confirming the two new tests fail (see below).
-
-**Priority 4 (the snapshot/worktree hint).** Handled as a consequence of both fixes rather than a
-separate branch: once a task resolves as `REPLIED` (via either fix), `abandon()` never calls
-`new Reply(Outcome.WORKER_FAILED, reason)` for that task at all, so the "the worker session was
-released before it replied; worktree=... branch=... snapshot=..." text is never constructed or
-attached to that ticket's outcome. It still appears, correctly, for a task that never got a reply
-(the existing `abandonFailsEveryPendingAsyncTicketForTheReleasedTarget` /
-`anAbandonedAsyncTaskPollsAsFailedNotPending` tests still pass unchanged).
-
-**Priority 2** was not needed — fix 1 makes `fleet_poll{ticket}` return the actual reply (the
-higher-priority option), so I did not fall back to "the ticket merely resolves as done with no
-content."
-
-## Tests — driven through the real delegation path, not the reply sink directly
-
-Both new tests in `fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java` go through
-`sendAsync` → `injectDelivery` → `ask` → `answer` (with a short timeout, so it genuinely times out,
-mirroring the ~25–120s real MCP-call bound vs. a longer resumed turn) → `reply` → `poll`/`abandon`.
-No test constructs a `Reply` and hands it to a sink directly.
-
-- `aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket` — asserts `fleet_poll{ticket}` (via
-  `messages.poll`) reaches `Phase.DONE` with the worker's actual reply text and
-  `replySource() == "reply"`, and that `hasStrandedReply(T)` stays `false` (proves the reply never
-  touched the inbox at all — fix 1 caught it).
-- `fleetStopAfterAnOrphanedReplyDoesNotFailTheTicket` — same setup, then calls `abandon(T, "the
-  worker session was released before it replied")` (what `fleet_stop` triggers) and asserts it
-  returns `false` (no failure recorded) and the ticket still polls `DONE` with the real reply.
-
-**Proof both fail without the change.** I temporarily short-circuited both new private methods
-(`askAnsweredAsyncTask` → always `null`, `recoverStrandedReply` → always `null`) — i.e. disabled
-both fixes — and ran just these two tests:
-
-```
-[ERROR] Tests run: 2, Failures: 2, Errors: 0, Skipped: 0
-dev.ltms.fleet.msg.MessageServiceTest.aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket
-  org.opentest4j.AssertionFailedError: expected: <DONE> but was: <PENDING>
-dev.ltms.fleet.msg.MessageServiceTest.fleetStopAfterAnOrphanedReplyDoesNotFailTheTicket
-  org.opentest4j.AssertionFailedError: a reply already arrived, so nothing here is a genuine failure
-  ==> expected: <false> but was: <true>
-```
-
-This is the exact bug: the ticket stays `PENDING` forever, and `abandon()` reports `true` (a
-failure) even though a reply had already arrived. I then restored both fixes (verified with
-`grep -n "TEMP #137-proof"` finding nothing) and re-ran — both pass.
-
-## Build
-
-Ran from `fleetd/`, unpiped, full output read (not `| tail`):
-
-```
-mvn clean install
-...
-[INFO] Tests run: 1039, Failures: 0, Errors: 0, Skipped: 0
-[INFO] BUILD SUCCESS
-[INFO] Total time:  36.315 s
-```
-
-Main was at 1037 tests; this branch adds the 2 new tests above → 1039, all green, `exit=0`.
-
-## What I could NOT check
-
-- No IDE tooling is mounted for me (worker), so no `ide_diagnostics`/IntelliJ inspection pass — only
-  `mvn clean install` (compiler + full test suite), as the worker procedure allows.
-- I cannot restart the daemon or dogfood this live — I have no forge/daemon control. This is
-  unverified against a real herdr pane, a real MCP client's ~60s call cap, or a real worker session;
-  everything above is verified only through the JUnit fixture's simulated timing
-  (`FakeHerdr`/`injector.onStatus`/direct `messages.answer(...,150)` calls), not a live fleet.
-  A primary should still consider a short live dogfood (an async delegation that asks, gets answered,
-  and takes longer than ~2 minutes to reply) before calling this closed.
-- I did not touch, and did not re-verify, the `fleet_ask` ~55s window itself (issue #61) — out of
-  scope per the brief.
-
-## Scope note (not investigated further)
-
-`answer()`'s nested/double-`fleet_ask` case (the worker asks a second question before ever
-replying to the first answer) has some pre-existing behaviour around which `turnId` a `QUESTION`
-resolution gets attributed to that I did not fully untangle — it predates this change, my fix does
-not touch it, and it is unrelated to the reported defect. Flagging only; not investigated further.
-
-## Handoff
-
-- Branch: `worker/cb-137-ask-ticket-e7760c-2`
-- Worktree root: `/Users/dai.ha/LTMS/.bridged-worktrees/734324-2`
-- Files changed:
-  - `fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java`
-  - `fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java`
-  - `REPORT-cb137.md` (this file)
-- Build: `Tests run: 1039, Failures: 0, Errors: 0, Skipped: 0` / `BUILD SUCCESS` (verbatim above)

fleetd/src/main/java/dev/ltms/fleet/member/HerdrPeerLauncher.java

@@ -112,6 +112,12 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      * daemon's own process is started the same way (a login shell sourcing the same secret store —
      * see CB-592's investigation of {@code secrets.sh}), so on a single-host deployment its env
      * mirrors what the pane's login shell is about to export.
+     *
+     * <p>fleetd #185 stage 2: that mirroring assumption holds only while the member pane runs under
+     * the SAME OS user as the daemon. When {@code memberHerdrSocket:} is configured, member panes
+     * run on a second herdr owned by a different user — different {@code $HOME}, different {@code
+     * secrets.sh}, different environment entirely — so this field's data no longer describes what a
+     * member pane inherits. See {@link #logCredentialGap} for how that mode is handled.
      */
     private final Supplier<Set<String>> hostEnvNames;
 
@@ -1230,6 +1236,67 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      */
     private final AtomicBoolean allowListGapLogged = new AtomicBoolean();
 
+    /**
+     * fleetd #185 stage 2: guards {@link #warnUnknownMemberEnvironment} to one WARN per launcher
+     * instance, not one per spawn — the same one-per-instance shape as {@link #unprotectedGapLogged}
+     * and {@link #allowListGapLogged}, kept as its own flag for the same reason those two are split:
+     * this mode is orthogonal to which of the other two branches would otherwise have fired.
+     */
+    private final AtomicBoolean unknownMemberEnvironmentWarned = new AtomicBoolean();
+
+    /**
+     * fleetd #185 stage 2: whether {@code memberHerdrSocket:} is configured, i.e. member panes run
+     * on a second herdr owned by a different OS user than the daemon's own process. Re-read from the
+     * live config on every call (same hot-reload shape as {@link #memberCredentials}), never cached,
+     * so a config reload takes effect on the next spawn without a restart.
+     *
+     * <p>{@link #config} is {@code null} on any call site that never threaded the full config
+     * through (every production {@code HerdrPeerLauncher} does; a handful of older tests do not) —
+     * treated the same as "not configured", which is the correct, permissive default: it is exactly
+     * today's single-daemon behaviour.
+     */
+    private boolean memberHerdrSocketConfigured() {
+        if (config == null) {
+            return false;
+        }
+        FleetConfig cfg = config.get();
+        return cfg != null && cfg.memberHerdrSocket() != null && !cfg.memberHerdrSocket().isBlank();
+    }
+
+    /**
+     * fleetd #185 stage 2: the single replacement WARN for {@link #logCredentialGap}'s usual
+     * conclusions when {@code memberHerdrSocket:} is configured. {@link #hostEnvNames} (and
+     * everything derived from it — {@code known}/{@code allow} coverage, the allow-list scrub's
+     * derived set) describes the DAEMON's own environment; under this config key member panes run as
+     * a different OS user with a different environment entirely, so neither "every member pane
+     * inherits them UNBLOCKED" nor "the scrub blanks them" is evidence-backed here — both would be
+     * reporting on the wrong process. Logged once, names the config key, and states the honest
+     * conclusion: the gap for member panes is UNKNOWN, not clean, so {@code memberCredentials} cannot
+     * be verified from this daemon. The one count it does report is scoped explicitly to fleetd's own
+     * environment, never presented as if it said anything about the member's — see {@link
+     * #logCredentialGap}'s javadoc for why this branch exists.
+     */
+    private void warnUnknownMemberEnvironment(FleetConfig.MemberCredentials creds) {
+        if (!unknownMemberEnvironmentWarned.compareAndSet(false, true)) {
+            return;
+        }
+        Set<String> covered = new HashSet<>(creds.known());
+        covered.addAll(creds.allow());
+        Set<String> hostNames = hostEnvNames.get();
+        long gapInFleetdsOwnEnv = hostNames.stream()
+                .filter(name -> CREDENTIAL_SHAPED_NAME.matcher(name).matches())
+                .filter(name -> !covered.contains(name))
+                .count();
+        log.warn("memberCredentials gap: memberHerdrSocket is configured, so member panes run under "
+                        + "a different OS user than fleetd's own process, with a different environment "
+                        + "entirely — fleetd has no channel to read that user's environment. {} of the "
+                        + "{} names in fleetd's OWN environment are credential-shaped and not on "
+                        + "known:/allow:, but that count describes fleetd's process, not the member "
+                        + "herdr's. The credential gap for member panes is UNKNOWN, not clean, and "
+                        + "memberCredentials cannot be verified from here.",
+                gapInFleetdsOwnEnv, hostNames.size());
+    }
+
     /**
      * CB-596 criterion 4: a credential-shaped host env var name on neither {@code known} nor
      * {@code allow} is not silently allowed — it is reported. {@link #hostEnvNames} enumerates the
@@ -1258,8 +1325,22 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
      * (same severity, and same guard, as the deny-by-default case — a name genuinely reaching a
      * member unprotected is equally serious whichever path put it there), and the names it says are
      * blanked keep the INFO.
+     *
+     * <p>fleetd #185 stage 2: everything above assumes the member pane runs under the same OS user
+     * as the daemon, so {@link #hostEnvNames} mirrors what the pane inherits — see that field's
+     * javadoc. When {@code memberHerdrSocket:} is configured that assumption is false: the member
+     * pane runs on a second herdr owned by a <em>different</em> user, and neither conclusion below
+     * ("inherits them UNBLOCKED" / "the scrub blanks them") is backed by evidence about that user's
+     * environment. So this method checks that first and, when configured, reports the honest
+     * "unknown, not clean" conclusion instead — see {@link #warnUnknownMemberEnvironment}. When
+     * {@code memberHerdrSocket:} is absent (the default, and the only mode this host runs) this
+     * branch is never taken and every line below is unchanged.
      */
     private void logCredentialGap(FleetConfig.MemberCredentials creds, Set<String> effectiveAllowed) {
+        if (memberHerdrSocketConfigured()) {
+            warnUnknownMemberEnvironment(creds);
+            return;
+        }
         Set<String> covered = new HashSet<>(creds.known());
         covered.addAll(creds.allow());
         List<String> gap = hostEnvNames.get().stream()

fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java

@@ -9,7 +9,6 @@ import dev.ltms.fleet.metrics.Metrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -168,12 +167,6 @@ public final class MessageService {
     private static final class Task {
         private final String ticket;
         private final String target;
-        /**
-         * When this task was created (#137 fix): the tiebreaker for which of several open tasks on
-         * one target gets a recovered reply in {@link #abandon} — the oldest, since it is the one
-         * that has been waiting longest.
-         */
-        private final long createdNanos;
         private final CompletableFuture<Reply> future = new CompletableFuture<>();
         /**
          * When {@link #future} resolved, or {@code null} while it is still pending — the clock
@@ -191,7 +184,6 @@ public final class MessageService {
         private Task(String ticket, String target, LongSupplier nowNanos) {
             this.ticket = ticket;
             this.target = target;
-            this.createdNanos = nowNanos.getAsLong();
             future.whenComplete((reply, ex) -> completedNanos = nowNanos.getAsLong());
         }
     }
@@ -383,19 +375,6 @@ 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.
-     *
      * @return always {@code true} — the reply resolved a live send, completed a parked ticket, or
      *         was queued
      */
@@ -413,21 +392,13 @@ public final class MessageService {
         // 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);
-            if (orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
-                if (orphan.turnId != null) {
-                    asyncTasksByTurn.remove(orphan.turnId, orphan);
-                }
-                count(FleetMetrics.REPLIES, "path", "async-recovered");
-                return true;                   // the ticket itself took it — no inbox stranding at all
+        Task orphan = askAnsweredAsyncTask(session);
+        if (orphan != null && orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
+            if (orphan.turnId != null) {
+                asyncTasksByTurn.remove(orphan.turnId, orphan);
             }
-        } else if (candidates.size() > 1) {
-            List<String> tickets = candidates.stream().map(t -> t.ticket).toList();
-            log.warn("reply from {} matches {} open async tickets {} — cannot tell which one it "
-                    + "answers, queuing to the inbox instead of guessing", session, candidates.size(),
-                    tickets);
+            count(FleetMetrics.REPLIES, "path", "async-recovered");
+            return true;                       // the ticket itself took it — no inbox stranding at all
         }
         inbox.publish(session, UUID.randomUUID().toString(), content);
         // CB-640: record the stranding itself (not just the reply text) so fleet health can see a
@@ -443,38 +414,21 @@ 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
+     * The 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). {@code null} 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}).
-     *
-     * <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).
      */
-    private List<Task> askAnsweredAsyncTasks(String target) {
-        List<Task> candidates = new ArrayList<>();
+    private Task askAnsweredAsyncTask(String target) {
         for (Task task : tasks.values()) {
             if (target.equals(task.target) && task.question == null && task.turnId != null
                     && !task.future.isDone()) {
-                candidates.add(task);
+                return task;
             }
         }
-        return candidates;
+        return null;
     }
 
     /** Record a counter sample when a registry is wired; a no-op in unit tests. */
@@ -519,55 +473,16 @@ public final class MessageService {
      * outcome is counted, so a torn-down delegation stops being invisible to {@code /metrics}.
      *
      * <p><strong>#137 defence in depth.</strong> {@link #reply} already hands a worker's real
-     * {@code fleet_reply} straight to the async ticket it belongs to whenever exactly one is still
-     * parked waiting for it (see {@link #askAnsweredAsyncTasks}), so by the time a session is
-     * released its tasks are normally already resolved — this loop's {@code complete} calls are then
-     * harmless no-ops (a {@link CompletableFuture} can only resolve once). But should some other path
-     * someday strand a reply in the inbox without completing its ticket, checking
+     * {@code fleet_reply} straight to the async ticket it belongs to whenever one is still parked
+     * waiting for it (see {@link #askAnsweredAsyncTask}), so by the time a session is released its
+     * tasks are normally already resolved — this loop's {@code complete} calls are then harmless
+     * no-ops (a {@link CompletableFuture} can only resolve once). But should some other path someday
+     * strand a reply in the inbox without completing its ticket, checking
      * {@link #hasStrandedReply(String)} here — before ever writing a failure — means a torn-down
      * session whose worker in fact replied is still reported {@code REPLIED} with that reply's own
      * text, never the misleading "the worker session was released before it replied" (which also
      * means the snapshot/worktree recovery hint that follows it never prints once a reply exists).
      *
-     * <p><strong>At most one task gets the recovered reply — and here, unlike {@link #reply}'s
-     * {@link #askAnsweredAsyncTasks}, {@code matching.size() >= 2} alone is reachable today.</strong>
-     * This method's {@code matching} filter has no {@code turnId != null} requirement, so it matches
-     * any plain (never-asked) open task too — and {@link #sendAsync} does not limit a target to one
-     * of those: a second {@code fleet_send{wait:false}} at a target that is still busy returns its own
-     * ticket immediately and simply parks its {@link #send} behind the target's session lock for up
-     * to {@link #ASYNC_TIMEOUT_MS}, exactly as {@code abandonFailsEveryPendingAsyncTicketForTheReleasedTarget}
-     * already proves. Before this fix, the loop below drained the strand once and then reused that
-     * same {@code Reply} for <em>every</em> task it walked past — so two open tasks really did both
-     * complete {@code REPLIED} with the same text (see the pre-fix loop in commit 97f6c33's parent).
-     * A stranded reply is one worker answer, so it can settle at most one open task on this target —
-     * never every open task, and never a guess. When more than one task is still open here, the
-     * recovered reply goes to the <em>oldest</em> (lowest {@link Task#createdNanos}) — it has been
-     * waiting longest, so it is the one most likely to be what the reply actually answers. Every
-     * other open task keeps the ordinary {@code WORKER_FAILED} path it would take without a stranded
-     * reply at all.
-     *
-     * <p><strong>{@code matching.size() >= 2} together with {@code hadStrandedReply} is a different
-     * question, and today it is defence in depth rather than a path this codebase's public API can
-     * drive.</strong> This class has exactly two sites that ever acquire a target's entry in
-     * {@code sessionLocks} — {@link #send} and {@link #answer} — and both open a {@link Rendezvous}
-     * waiter for that same target as the very first thing they do after acquiring the lock, then hold
-     * lock and waiter together for the rest of their critical section ({@link #send} also clears
-     * {@link #strandedReplies} right there, the instant it opens its waiter — before it ever enqueues
-     * delivery). So "the session lock is held" and "a live waiter is open for it" are the same fact
-     * throughout this class, and {@link #reply}'s fast path always resolves a currently-open waiter
-     * directly rather than stranding. The two facts this method wants therefore cannot be produced
-     * side by side: while the lock is held, a real reply resolves the open waiter directly and never
-     * reaches {@link #strandedReplies}; the instant the lock is free, any parked matching task's own
-     * {@link #send} that is scheduled next wins it and, by opening its waiter, clears the strand again
-     * before this method ever runs. There is no way to hold that lock open-but-unaccepted from outside
-     * {@link #send}/{@link #answer} to freeze a window in between. Constructing both facts at once
-     * through {@code sendAsync}/{@code reply}/{@code ask}/{@code answer} would need a race against
-     * virtual-thread scheduling, not a deterministic sequence — so the oldest-wins code below stays as
-     * defence in depth against a regression to that mechanism (e.g. clearing {@link #strandedReplies}
-     * on a narrower condition than "any acceptance"), not because today's test suite exercises the
-     * conjunction. {@code matching.size() >= 2} alone, without a strand, is exactly what
-     * {@code abandonFailsEveryPendingAsyncTicketForTheReleasedTarget} already covers.
-     *
      * @return true if a live waiter or an async task was failed (never true for one recovered as a
      *         reply — see the note above)
      */
@@ -579,39 +494,21 @@ public final class MessageService {
         CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
         boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason);
         boolean asyncFailed = false;
-
-        List<Task> matching = new ArrayList<>();
+        Reply recovered = null; // lazily drained at most once, only if a task actually needs it
         for (Task task : tasks.values()) {
-            if (target.equals(task.target) && task.question == null && !task.future.isDone()) {
-                matching.add(task);
+            if (!target.equals(task.target) || task.question != null || task.future.isDone()) {
+                continue;
             }
-        }
-        Task recoveryTask = null;
-        if (hadStrandedReply && !matching.isEmpty()) {
-            recoveryTask = matching.get(0);
-            for (Task candidate : matching) {
-                if (candidate.createdNanos < recoveryTask.createdNanos) {
-                    recoveryTask = candidate;
-                }
+            if (hadStrandedReply && recovered == null) {
+                recovered = recoverStrandedReply(target);
             }
-        }
-        Reply recovered = recoveryTask != null ? recoverStrandedReply(target) : null;
-        for (Task task : matching) {
-            boolean isRecovery = task == recoveryTask && recovered != null;
-            Reply outcome = isRecovery ? recovered : new Reply(Outcome.WORKER_FAILED, reason);
+            Reply outcome = recovered != null ? recovered : new Reply(Outcome.WORKER_FAILED, reason);
             if (task.future.complete(outcome)) {
                 if (outcome.outcome() == Outcome.WORKER_FAILED) {
                     asyncFailed = true;
                 } else if (task.turnId != null) {
                     asyncTasksByTurn.remove(task.turnId, task);
                 }
-            } else if (isRecovery) {
-                // The recovered reply was already drained out of the inbox, but this task resolved
-                // through another path (e.g. a concurrent reply() or a second abandon() racing this
-                // one) between us choosing it and completing it here. Put the reply back rather than
-                // lose it silently — it may still belong to some other still-open task, or the next
-                // caller that drains this target's inbox.
-                inbox.publish(target, UUID.randomUUID().toString(), recovered.text());
             }
         }
         if (failed) {
@@ -626,10 +523,6 @@ public final class MessageService {
      * stranding fact raced away, e.g. a lead's own {@code fleet_poll} on the raw session already
      * drained it first). When more than one message is queued, only the newest is the worker's actual
      * final answer ({@link #drainReplies} returns them oldest-first).
-     *
-     * <p>This does drain (removes the messages from the inbox) before the caller knows whether the
-     * task it is recovering for will actually accept them — {@link #abandon} is the one that puts a
-     * reply back if its {@code complete} call turns out to lose the race.
      */
     private Reply recoverStrandedReply(String target) {
         var messages = drainReplies(target);

fleetd/src/test/java/dev/ltms/fleet/member/HerdrPeerLauncherAllowListWiringTest.java

@@ -257,6 +257,137 @@ class HerdrPeerLauncherAllowListWiringTest {
                         + appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList());
     }
 
+    /**
+     * fleetd #185 stage 2 pin: with {@code memberHerdrSocket:} absent (today's only mode, and the
+     * default — this host runs no other), the gap detector's WARN/INFO conclusions read exactly as
+     * they did before this fix. Real path: {@code FLEETD_WORKER_TOKEN} (the test profile's own
+     * {@code tokenEnv}) is a name the derived allow-list keeps, so it gets the "UNBLOCKED" WARN;
+     * {@code SOME_UNKNOWN_SECRET_TOKEN} is not derived from anywhere, so it gets the "scrub blanks
+     * them" INFO. This is the exact shape #185 stage 2 must not touch on this path.
+     */
+    @Test
+    void gapConclusionsAreByteIdenticalWhenMemberHerdrSocketIsAbsent() {
+        FakeHerdr herdr = new FakeHerdr();
+        Set<String> hostEnvNames = Set.of("FLEETD_WORKER_TOKEN", "SOME_UNKNOWN_SECRET_TOKEN");
+        WiringLauncher launcher = new WiringLauncher(herdr, allowList(), "/bin/zsh", () -> hostEnvNames,
+                () -> config(null));
+
+        List<String> messages = spawnAndCaptureLogs(launcher);
+
+        assertTrue(messages.contains("memberCredentials gap: 1 credential-shaped env var name(s) are on "
+                        + "neither known: nor allow: — the derived allow-list keeps them anyway (a "
+                        + "profile's gitTokenEnv/gitHostEnv/tokenEnv/env: names one, or this spawn "
+                        + "injects it), so every member pane inherits them UNBLOCKED — [FLEETD_WORKER_TOKEN]. "
+                        + "Add each to memberCredentials.known (or .allow if a member legitimately needs "
+                        + "it), or remove it from whatever profile setting derives it in."),
+                "expected the pre-existing UNBLOCKED WARN unchanged, got: " + messages);
+        assertTrue(messages.contains("memberCredentials gap: 1 credential-shaped env var name(s) are on "
+                        + "neither known: nor allow: — [SOME_UNKNOWN_SECRET_TOKEN]. The allow-list scrub "
+                        + "blanks them anyway (they are not on the derived allow-list), so no member pane "
+                        + "keeps them; add each to memberCredentials.known or .allow to make that explicit."),
+                "expected the pre-existing 'scrub blanks them' INFO unchanged, got: " + messages);
+    }
+
+    /**
+     * fleetd #185 stage 2: with {@code memberHerdrSocket:} configured, member panes run under a
+     * different OS user — {@link HerdrPeerLauncher#hostEnvNames} describes fleetd's own process, not
+     * that user's. Neither "inherits them UNBLOCKED" nor "scrub blanks them" is evidence-backed
+     * there, so neither may print; the single unknown-environment WARN must, naming the config key.
+     */
+    @Test
+    void gapDetectorReportsUnknownInsteadOfAConclusionWhenMemberHerdrSocketIsConfigured() {
+        FakeHerdr herdr = new FakeHerdr();
+        Set<String> hostEnvNames = Set.of("FLEETD_WORKER_TOKEN", "SOME_UNKNOWN_SECRET_TOKEN");
+        WiringLauncher launcher = new WiringLauncher(herdr, allowList(), "/bin/zsh", () -> hostEnvNames,
+                () -> configWithMemberHerdrSocket("/tmp/other-user-herdr.sock"));
+
+        List<String> messages = spawnAndCaptureLogs(launcher);
+
+        assertTrue(messages.stream().anyMatch(m -> m.contains("memberHerdrSocket")
+                        && m.contains("UNKNOWN") && m.contains("cannot be verified")),
+                "expected the unknown-member-environment WARN naming memberHerdrSocket, got: " + messages);
+        assertFalse(messages.stream().anyMatch(m -> m.contains("UNBLOCKED")),
+                "the 'inherits them UNBLOCKED' conclusion must not print once the evidence is about "
+                        + "the wrong (daemon's own) environment — got: " + messages);
+        assertFalse(messages.stream().anyMatch(m -> m.contains("scrub blanks them")),
+                "the 'scrub blanks them' conclusion must not print once the evidence is about the "
+                        + "wrong (daemon's own) environment — got: " + messages);
+    }
+
+    /**
+     * fleetd #185 stage 2: the unknown-environment WARN is a standing fact about this launcher's
+     * configuration, not per-spawn news — it must fire once per launcher instance, the same shape as
+     * every other one-time WARN in this class (e.g. {@code warnNonZsh}).
+     */
+    @Test
+    void theUnknownEnvironmentWarnFiresOnceNotOncePerSpawn() {
+        FakeHerdr herdr = new FakeHerdr();
+        Set<String> hostEnvNames = Set.of("FLEETD_WORKER_TOKEN", "SOME_UNKNOWN_SECRET_TOKEN");
+        WiringLauncher launcher = new WiringLauncher(herdr, allowList(), "/bin/zsh", () -> hostEnvNames,
+                () -> configWithMemberHerdrSocket("/tmp/other-user-herdr.sock"));
+
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        Level original = logger.getLevel();
+        logger.setLevel(Level.WARN);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(original);
+        }
+
+        long count = appender.list.stream()
+                .filter(e -> e.getFormattedMessage().contains("cannot be verified from here"))
+                .count();
+        assertEquals(1, count, "the unknown-member-environment WARN must fire once per launcher "
+                + "instance, not once per spawn — got " + count + " occurrence(s) among: "
+                + appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList());
+    }
+
+    /**
+     * Hard constraint: the gap detector must never log an env var VALUE, only its NAME. {@code
+     * SOME_UNKNOWN_SECRET_TOKEN} resolves to a distinctive canary value through the same {@code env}
+     * lookup the launcher uses elsewhere (SHELL, PATH, token resolution) — proving the value IS
+     * resolvable does not mean the detector reads it, since {@link HerdrPeerLauncher#hostEnvNames}
+     * (names only) is its data source, never {@code env.apply(name)} for those names.
+     */
+    @Test
+    void theGapDetectorNeverLogsAnEnvVarValueOnlyItsName() {
+        FakeHerdr herdr = new FakeHerdr();
+        String canary = "sekrit-value-CANARY-9f3a1b7c";
+        Set<String> hostEnvNames = Set.of("FLEETD_WORKER_TOKEN", "SOME_UNKNOWN_SECRET_TOKEN");
+        WiringLauncher launcher = new WiringLauncher(herdr, allowList(), "/bin/zsh", () -> hostEnvNames,
+                () -> config(null), Map.of("SOME_UNKNOWN_SECRET_TOKEN", canary));
+
+        List<String> messages = spawnAndCaptureLogs(launcher);
+
+        assertTrue(messages.stream().anyMatch(m -> m.contains("SOME_UNKNOWN_SECRET_TOKEN")),
+                "expected the credential-shaped NAME to appear in the log, got: " + messages);
+        assertFalse(messages.stream().anyMatch(m -> m.contains(canary)),
+                "the log must never contain an env var VALUE, only its NAME — got: " + messages);
+    }
+
+    /** Spawn once through the real launcher path, capturing every INFO+ line this class logs. */
+    private static List<String> spawnAndCaptureLogs(HerdrPeerLauncher launcher) {
+        Logger logger = (Logger) LoggerFactory.getLogger(HerdrPeerLauncher.class);
+        Level original = logger.getLevel();
+        logger.setLevel(Level.INFO);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            launcher.spawn(new SpawnRequest("test", null, null, null, null, MemberRole.DEV));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(original);
+        }
+        return appender.list.stream().map(ILoggingEvent::getFormattedMessage).toList();
+    }
+
     private static String readAll(Path p) {
         try {
             return Files.readString(p);
@@ -294,9 +425,24 @@ class HerdrPeerLauncherAllowListWiringTest {
 
         WiringLauncher(FakeHerdr herdr, Supplier<FleetConfig.MemberCredentials> creds, String shell,
                        Supplier<Set<String>> hostEnvNames, Supplier<FleetConfig> config) {
+            this(herdr, creds, shell, hostEnvNames, config, Map.of());
+        }
+
+        /**
+         * Plus a host-env value map (name → value), resolved through the same {@code env} lookup
+         * every adapter uses for {@code SHELL}/{@code PATH}/token resolution — fleetd #185 stage 2's
+         * "the gap detector never logs a value" tests use this to prove a value that IS resolvable
+         * for a credential-shaped name never reaches the log, since the detector only ever reads
+         * {@code hostEnvNames} (names), never {@code env.apply(name)} (values), for those names.
+         */
+        WiringLauncher(FakeHerdr herdr, Supplier<FleetConfig.MemberCredentials> creds, String shell,
+                       Supplier<Set<String>> hostEnvNames, Supplier<FleetConfig> config,
+                       Map<String, String> extraEnvValues) {
             super("test", new AgentControl(herdr), new WorkspaceControl(herdr),
                     Map.of("test", profile()), "test",
-                    name -> "SHELL".equals(name) ? shell : null,
+                    name -> "SHELL".equals(name) ? shell
+                            : (extraEnvValues != null && extraEnvValues.containsKey(name))
+                                    ? extraEnvValues.get(name) : null,
                     0, () -> 0L, () -> { }, null, creds, hostEnvNames, config);
         }
 
@@ -321,6 +467,16 @@ class HerdrPeerLauncherAllowListWiringTest {
                 null, null, null, null, null).withDefaults();
     }
 
+    /**
+     * fleetd #185 stage 2: a config with {@code memberHerdrSocket:} set — member panes run on a
+     * second herdr owned by a different OS user, so {@link HerdrPeerLauncher#hostEnvNames} no
+     * longer describes what a member pane inherits.
+     */
+    private static FleetConfig configWithMemberHerdrSocket(String memberHerdrSocket) {
+        return new FleetConfig(null, null, memberHerdrSocket, Map.of(), null, null, null, null, null,
+                null, null, null, null, null, null, null, null, null, null, null).withDefaults();
+    }
+
     /** The generated directory is a temp directory; make sure the test does not leave a pile. */
     @Test
     void theGeneratedDirectoryIsRemovedWhenThePaneIsStopped() {

fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java

@@ -688,32 +688,6 @@ class MessageServiceTest {
         assertFailedTicket(third, "agent target term_a not found");
     }
 
-    // --- #137 follow-up: abandon() must not guess when more than one task is open ---------------
-    //
-    // A test combining a genuine stranded reply (hasStrandedReply(T)==true) with two simultaneously
-    // open matching tasks was attempted here and removed after investigation showed the combination
-    // is not reachable through the public API today, not merely hard to time right:
-    //
-    // This class has exactly two call sites that ever hold a target's entry in the session-lock map
-    // (send() and answer()), and both open a Rendezvous waiter for that same target as the first thing
-    // they do after acquiring the lock, holding lock and waiter together for their whole critical
-    // section. So "the lock is held" and "a live waiter is open" are the same fact throughout this
-    // class. reply()'s fast path always resolves a currently-open waiter directly instead of
-    // stranding — so a strand can only be created while NO task is accepted (lock free), and the
-    // instant the lock is next taken (by any parked matching task's own send(), the moment it is
-    // scheduled), that acceptance clears strandedReplies again (see send()'s CB-640 comment) before
-    // abandon() can ever observe both facts together. Confirmed empirically too: an earlier version of
-    // this test stranded a reply, then created an "accepted" task (awaitWaiting()) followed by a
-    // "parked" one — and the accepted task's own acceptance silently cleared the strand it was
-    // supposed to be racing against, so the parked task came back WORKER_FAILED instead of DONE, not
-    // because the fix was missing but because the test's premise could not be constructed.
-    //
-    // The reachable half — matching.size() >= 2 alone, no strand — is exactly what
-    // abandonFailsEveryPendingAsyncTicketForTheReleasedTarget already covers (all fail, none guess).
-    // The oldest-wins code in abandon() stays as defence in depth (see its own javadoc) against a
-    // regression that would make the conjunction reachable, e.g. clearing strandedReplies on a
-    // narrower condition than "any acceptance" — not because this suite exercises it today.
-
     @Test
     void abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer() throws Exception {
         String ticket = messages.sendAsync(T, "task that asks");