#137 follow-up: document defect-1 and defect-2 reachability, drop unconstructible test

Defect 1 (reply()'s askAnsweredAsyncTasks returning >1 candidate): confirmed
unreachable today. Documented why in three places — hasAsyncQuestion matches
any task with a stamped turnId (not just an open question), and answer()'s
clearAsyncQuestion(turnId, false) leaves that stamp in place until the
resumed turn's own future resolves — so a second task can never reach the
same eligible state while a first one holds it. Kept the defensive
inbox-fallback branch as defence in depth against that guarantee weakening,
per review instruction; no test seam added.

Defect 2 (abandon()'s broader `matching` filter applying a stranded reply to
more than one task): matching.size() >= 2 alone IS reachable (already
covered by abandonFailsEveryPendingAsyncTicketForTheReleasedTarget) and was
a real pre-fix bug (97f6c33's parent reused one drained reply for every
matching task). But hadStrandedReply == true together with matching.size()
>= 2, at the instant abandon() runs, is not constructible through the public
API: send() and answer() are the only two sites that ever hold a target's
session lock, and both open a Rendezvous waiter for that target as the first
thing they do while holding it — so "lock held" and "waiter open" are the
same fact throughout this class, and reply()'s fast path always resolves an
open waiter directly instead of stranding. A strand can only be created
while no task is accepted, and the moment the lock is next taken, that
acceptance clears the strand again before abandon() can observe both facts
together. Documented this in abandon()'s javadoc and removed the earlier
attempt at a deterministic test for the conjunction, whose apparent failure
was an invalid premise (the "accepted" task's own acceptance silently
cleared the strand it was meant to race against), not the fix being absent.

REPORT-cb137.md

@@ -0,0 +1,166 @@
+# 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/fleetd.example.yaml

@@ -634,6 +634,17 @@ guard:
 # to a sibling directory of the repo root.
 # worktreeRoot: /Users/me/src/.bridged-worktrees
 
+# Worktree group sharing (fleetd #185 stage 3). OPTIONAL, off by default. Names an OS group
+# that a provisioned worktree's repo is made group-writable for (git config
+# core.sharedRepository group, plus a one-time chgrp/chmod/setgid fix-up), so a member spawned
+# under a DIFFERENT OS user (see memberHerdrSocket) can write its own worktree, its
+# per-worktree git metadata, and its own commit objects — without it, every file GitWorktrees
+# creates is owned by fleetd's own uid and unwritable by another user.
+# CAUTION: this isolates credentials, not the repository — a member in the group can still
+# write the operator's git objects and refs in the shared repo. The operator running fleetd
+# must already be a member of the named group, or every provisioning spawn fails loudly.
+# worktreeGroup: fleet-workers
+
 # Session lifecycle limits (CB-303). All knobs are opt-in; omit or set to null to keep
 # the feature disabled. By default the daemon never reaps, caps, or drains sessions.
 #   idleTtlSeconds     → reap READY/DONE sessions idle longer than this (never BUSY/SPAWNING)

fleetd/src/main/java/dev/ltms/fleet/Fleetd.java

@@ -224,7 +224,7 @@ public final class Fleetd {
             contextCap = cfg.lifecycle().contextCap();
         }
         boolean clearAfterTurn = cfg.lifecycle() != null && cfg.lifecycle().clearAfterTurn();
-        SessionManager sessions = new SessionManager(workers, new GitWorktrees(cfg.worktreeRoot()),
+        SessionManager sessions = new SessionManager(workers, new GitWorktrees(cfg.worktreeRoot(), cfg.worktreeGroup()),
                 System::nanoTime, contextCap, clearAfterTurn);
         liveCountRef.set(profileName -> (int) sessions.roster().stream()
                 .filter(s -> profileName.equals(s.profile()))

fleetd/src/main/java/dev/ltms/fleet/config/FleetConfig.java

@@ -76,6 +76,14 @@ import java.util.Set;
  *                      stay on {@code broker}'s vhost). {@code null} → no lead mailbox is opened.
  *                      Config parsing + accessors only — nothing here wires it into a live
  *                      {@code LeadMailbox}; that is a separate ticket. See {@link Coordinator}.
+ * @param worktreeGroup optional OS group name (fleetd #185 stage 3) that makes a provisioned
+ *                      worktree's repo group-shared, so a member running as a different OS user
+ *                      (see {@code memberHerdrSocket}) can write its own worktree, its per-worktree
+ *                      git metadata, and its own commit objects. {@code null}/blank/empty ⇒ off,
+ *                      today's behaviour unchanged (every file stays owned by fleetd's own uid).
+ *                      <strong>This isolates credentials, not the repository</strong>: a member in
+ *                      the group can still write the operator's git objects and refs in the shared
+ *                      repo. See {@link dev.ltms.fleet.session.Worktrees#shareWithGroup}.
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public record FleetConfig(
@@ -98,7 +106,20 @@ public record FleetConfig(
         ConfigReload configReload,
         Integer quarantineCooldownSeconds,
         MemberCredentials memberCredentials,
-        Coordinator coordinator) {
+        Coordinator coordinator,
+        String worktreeGroup) {
+
+    /** Back-compat form before the {@code worktreeGroup} key was added. */
+    public FleetConfig(Bind bind, String herdrSocket, String memberHerdrSocket, Map<String, Profile> profiles,
+                         Guard guard, String worktreeRoot, Lifecycle lifecycle, Integer spawnReadyTimeoutMs,
+                         Integer spawnReadyPollMs, Broker broker, Primary primary, Fleet fleet,
+                         LeadHeartbeat leadHeartbeat, Health health, String placement, Auth auth,
+                         ConfigReload configReload, Integer quarantineCooldownSeconds,
+                         MemberCredentials memberCredentials, Coordinator coordinator) {
+        this(bind, herdrSocket, memberHerdrSocket, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs,
+                spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, health, placement, auth,
+                configReload, quarantineCooldownSeconds, memberCredentials, coordinator, null);
+    }
 
     /** Back-compat form before the {@code coordinator:} block was added. */
     public FleetConfig(Bind bind, String herdrSocket, Map<String, Profile> profiles, Guard guard,
@@ -109,7 +130,7 @@ public record FleetConfig(
                          MemberCredentials memberCredentials) {
         this(bind, herdrSocket, null, profiles, guard, worktreeRoot, lifecycle, spawnReadyTimeoutMs,
                 spawnReadyPollMs, broker, primary, fleet, leadHeartbeat, health, placement, auth,
-                configReload, quarantineCooldownSeconds, memberCredentials, null);
+                configReload, quarantineCooldownSeconds, memberCredentials, null, null);
     }
 
     /** Back-compat form before the CB-596 {@code memberCredentials:} block was added. */
@@ -1325,7 +1346,7 @@ public record FleetConfig(
             "bind", "herdrSocket", "memberHerdrSocket", "profiles", "guard", "worktreeRoot",
             "lifecycle", "spawnReadyTimeoutMs", "spawnReadyPollMs", "broker", "primary", "fleet",
             "leadHeartbeat", "health", "placement", "auth", "configReload", "quarantineCooldownSeconds",
-            "memberCredentials", "coordinator");
+            "memberCredentials", "coordinator", "worktreeGroup");
 
     /** Load and validate config from {@code path}. */
     public static FleetConfig load(Path path) {
@@ -1943,9 +1964,11 @@ public record FleetConfig(
                 : new MemberCredentials(null, List.of(), List.of());
         // coordinator is left as-is, like broker/primary above: null keeps no LeadMailbox opened,
         // and this ticket's Coordinator is config-only anyway (nothing yet reads it at startup).
+        // worktreeGroup is left as-is (fleetd #185 stage 3): null/blank is "off", and there is no
+        // sane non-null default — an OS group name is operator-specific.
         return new FleetConfig(b, herdrSocket, memberHerdrSocket, profiles, g, worktreeRoot, l, timeout, pollMs,
                 broker, primary, f, leadHeartbeat, health, placementOrDefault, a, configReload,
-                quarantineCooldown, mc, coordinator);
+                quarantineCooldown, mc, coordinator, worktreeGroup);
     }
 
     /**

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

@@ -9,6 +9,7 @@ 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;
@@ -167,6 +168,12 @@ 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
@@ -184,6 +191,7 @@ 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());
         }
     }
@@ -375,6 +383,19 @@ 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
      */
@@ -392,13 +413,21 @@ 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.
-        Task orphan = askAnsweredAsyncTask(session);
-        if (orphan != null && orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
-            if (orphan.turnId != null) {
-                asyncTasksByTurn.remove(orphan.turnId, orphan);
+        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
             }
-            count(FleetMetrics.REPLIES, "path", "async-recovered");
-            return true;                       // the ticket itself took it — no inbox stranding at all
+        } 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);
         }
         inbox.publish(session, UUID.randomUUID().toString(), content);
         // CB-640: record the stranding itself (not just the reply text) so fleet health can see a
@@ -414,21 +443,38 @@ public final class MessageService {
     }
 
     /**
-     * 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
+     * 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}).
+     *
+     * <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 Task askAnsweredAsyncTask(String target) {
+    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()) {
-                return task;
+                candidates.add(task);
             }
         }
-        return null;
+        return candidates;
     }
 
     /** Record a counter sample when a registry is wired; a no-op in unit tests. */
@@ -473,16 +519,55 @@ 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 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
+     * {@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
      * {@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)
      */
@@ -494,21 +579,39 @@ public final class MessageService {
         CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
         boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason);
         boolean asyncFailed = false;
-        Reply recovered = null; // lazily drained at most once, only if a task actually needs it
+
+        List<Task> matching = new ArrayList<>();
         for (Task task : tasks.values()) {
-            if (!target.equals(task.target) || task.question != null || task.future.isDone()) {
-                continue;
+            if (target.equals(task.target) && task.question == null && !task.future.isDone()) {
+                matching.add(task);
             }
-            if (hadStrandedReply && recovered == null) {
-                recovered = recoverStrandedReply(target);
+        }
+        Task recoveryTask = null;
+        if (hadStrandedReply && !matching.isEmpty()) {
+            recoveryTask = matching.get(0);
+            for (Task candidate : matching) {
+                if (candidate.createdNanos < recoveryTask.createdNanos) {
+                    recoveryTask = candidate;
+                }
             }
-            Reply outcome = recovered != null ? recovered : new Reply(Outcome.WORKER_FAILED, reason);
+        }
+        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);
             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) {
@@ -523,6 +626,10 @@ 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/main/java/dev/ltms/fleet/session/GitWorktrees.java

@@ -23,6 +23,7 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
@@ -88,24 +89,64 @@ public final class GitWorktrees implements Worktrees {
     );
 
     private final String configuredRoot;
+    /** OS group name for {@link #shareWithGroup} (fleetd #185 stage 3); {@code null} ⇒ feature off. */
+    private final String group;
     private final Consumer<String> afterWorktreeAdded;
+    /** 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
+     *  group runs, without a real second OS user or OS group on this host. */
+    private final Function<String[], String> shareGroupRunner;
     private final SecureRandom random = new SecureRandom();
     private final AtomicLong seq = new AtomicLong();
 
     /** Default constructor: worktree root is derived per-repo as {@code <repoRoot>/../.bridged-worktrees}. */
     public GitWorktrees() {
-        this(null);
+        this(null, (String) null);
     }
 
-    /** @param configuredRoot nullable absolute or relative path; null/blank derives a sibling of the repo root. */
+    /**
+     * @param configuredRoot nullable absolute or relative path; null/blank derives a sibling of
+     *                       the repo root. No {@code worktreeGroup} configured — {@link #shareWithGroup}
+     *                       is a no-op.
+     */
     public GitWorktrees(String configuredRoot) {
-        this(configuredRoot, _ -> {});
+        this(configuredRoot, (String) null);
+    }
+
+    /**
+     * @param configuredRoot nullable absolute or relative path; null/blank derives a sibling of
+     *                       the repo root.
+     * @param group          optional OS group name (fleetd #185 stage 3, {@code worktreeGroup:} in
+     *                       config); null/blank ⇒ {@link #shareWithGroup} is a no-op.
+     */
+    public GitWorktrees(String configuredRoot, String group) {
+        this(configuredRoot, group, _ -> {});
     }
 
     /** Test seam for changing a real worktree between its creation and its security check. */
     GitWorktrees(String configuredRoot, Consumer<String> afterWorktreeAdded) {
+        this(configuredRoot, null, afterWorktreeAdded);
+    }
+
+    /** Test seam combining a configurable {@code group} with {@link #afterWorktreeAdded}. */
+    GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded) {
+        this(configuredRoot, group, afterWorktreeAdded, null);
+    }
+
+    /**
+     * Full test seam: also overrides how {@link #shareWithGroup}'s processes run (fleetd #185
+     * stage 3), so a unit test can prove "no group configured ⇒ no process spawned" and inspect
+     * exactly what commands a configured group runs, without a real second OS user/group.
+     *
+     * @param shareGroupRunner {@code null} ⇒ the real {@link #exec(String...)}.
+     */
+    GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded,
+                 Function<String[], String> shareGroupRunner) {
         this.configuredRoot = configuredRoot;
+        this.group = (group == null || group.isBlank()) ? null : group;
         this.afterWorktreeAdded = afterWorktreeAdded == null ? _ -> {} : afterWorktreeAdded;
+        this.shareGroupRunner = shareGroupRunner != null ? shareGroupRunner : this::exec;
     }
 
     @Override
@@ -607,6 +648,126 @@ public final class GitWorktrees implements Worktrees {
         return deleted;
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * <p>fleetd #185 stage 3. No-op — no process spawned, nothing logged — when {@link #group} is
+     * null/blank. Otherwise:
+     * <ol>
+     *   <li>{@code git -C repoRoot config core.sharedRepository group} so every future write by
+     *       either uid stays group-writable;</li>
+     *   <li>a one-time {@code chgrp}/{@code chmod g+rwX} fix-up over the worktree directory and,
+     *       under the repo's <em>common</em> git directory, {@code objects}, {@code refs},
+     *       {@code logs}, {@code worktrees} and {@code packed-refs} — with setgid
+     *       ({@code chmod g+s}) applied only to the directories among them, so files created later
+     *       inherit the group;</li>
+     *   <li>one INFO line naming the group and the paths touched.</li>
+     * </ol>
+     *
+     * <p><b>Every path is skipped when it does not exist.</b> {@code .git/logs} is absent in a repo
+     * with {@code core.logAllRefUpdates=false} or one that has had no ref update yet, and
+     * {@code packed-refs} is absent until refs are packed. Passing a missing path to {@code chgrp}
+     * exits non-zero, which would fail <em>every</em> provisioning spawn with a message blaming a
+     * group that is in fact fine.
+     *
+     * <p><b>The git directory is resolved, not assumed.</b> {@code <repoRoot>/.git} is a
+     * <em>file</em>, not a directory, when the checkout is itself a linked worktree — the very
+     * thing this class creates for every member. {@code git rev-parse --git-common-dir} gives the
+     * real shared store, and it may answer relatively, so it is resolved against {@code repoRoot}.
+     *
+     * <p><b>The fix-up re-runs on every spawn, by design.</b> {@code core.sharedRepository=group}
+     * governs only what git writes <em>after</em> it is set; the walk is what covers everything
+     * already on disk. It is not redundant work to optimise away — dropping it silently leaves
+     * pre-existing objects unreadable to the member. It costs three walks of the object store per
+     * spawn (about 3000 files in this repo, well under a second, but it grows with the repo).
+     *
+     * <p>This only fixes up file ownership/permissions on the operator's shared repo so a
+     * different-uid member can write to it — it isolates credentials, not the repository. A member
+     * in the group can still write the operator's git objects and refs.
+     *
+     * <p>Fails loudly: a missing group, or a {@code chgrp}/{@code chmod} refused because the
+     * operator is not a member of it, becomes a {@link WorktreeException} naming the group — never
+     * a silent skip that leaves a member unable to work with nothing in the log to explain why.
+     */
+    @Override
+    public void shareWithGroup(String repoRoot, String worktreePath) {
+        if (group == null) {
+            return;
+        }
+        List<String> touched = new ArrayList<>();
+        try {
+            shareGroupRunner.apply(new String[]{"git", "-C", repoRoot, "config", "core.sharedRepository", "group"});
+            String commonDir = gitCommonDir(repoRoot);
+            shareGroupPathIfPresent(worktreePath, true, touched);
+            for (String name : List.of("objects", "refs", "logs", "worktrees")) {
+                shareGroupPathIfPresent(commonDir + "/" + name, true, touched);
+            }
+            shareGroupPathIfPresent(commonDir + "/packed-refs", false, touched);
+        } catch (WorktreeException e) {
+            throw new WorktreeException("cannot share worktree with group '" + group + "': "
+                    + e.getMessage() + " — the group must exist, and the fleetd operator ("
+                    + System.getProperty("user.name") + ") must be a member of it", e);
+        }
+        log.info("worktreeGroup={} shared repoRoot={} worktreePath={} paths={}",
+                group, repoRoot, worktreePath, touched);
+    }
+
+    /**
+     * The repo's <em>common</em> git directory as an absolute path — where {@code objects},
+     * {@code refs} and {@code worktrees} actually live. {@code git rev-parse --git-common-dir}
+     * answers relative to {@code repoRoot} in the ordinary case ({@code .git}) and absolutely for a
+     * linked worktree, so the answer is resolved against {@code repoRoot} either way. Never
+     * hardcode {@code repoRoot + "/.git"}: that is a FILE when the checkout is itself a linked
+     * worktree.
+     */
+    private String gitCommonDir(String repoRoot) {
+        String answer = shareGroupRunner.apply(
+                new String[]{"git", "-C", repoRoot, "rev-parse", "--git-common-dir"});
+        String trimmed = answer == null ? "" : answer.trim();
+        if (trimmed.isEmpty()) {
+            trimmed = ".git";
+        }
+        return Path.of(repoRoot).resolve(trimmed).normalize().toString();
+    }
+
+    /**
+     * {@link #shareGroupPath} when {@code path} exists, recording it in {@code touched}; otherwise
+     * nothing at all. A missing path is normal, not an error — see {@link #shareWithGroup}'s
+     * javadoc for which ones are routinely absent and why passing them to {@code chgrp} would fail
+     * every spawn.
+     */
+    private void shareGroupPathIfPresent(String path, boolean recursive, List<String> touched) {
+        if (!Files.exists(Path.of(path))) {
+            return;
+        }
+        shareGroupPath(path, recursive);
+        touched.add(path);
+    }
+
+    /**
+     * {@code chgrp}/{@code chmod g+rwX} {@code path} to {@link #group}. When {@code recursive},
+     * also walks the directories under {@code path} (including {@code path} itself, when it is a
+     * directory) and sets setgid on each — directories only, per the javadoc on
+     * {@link #shareWithGroup}.
+     */
+    private void shareGroupPath(String path, boolean recursive) {
+        List<String> chgrp = new ArrayList<>(List.of("chgrp"));
+        if (recursive) chgrp.add("-R");
+        chgrp.add(group);
+        chgrp.add(path);
+        shareGroupRunner.apply(chgrp.toArray(new String[0]));
+
+        List<String> chmod = new ArrayList<>(List.of("chmod"));
+        if (recursive) chmod.add("-R");
+        chmod.add("g+rwX");
+        chmod.add(path);
+        shareGroupRunner.apply(chmod.toArray(new String[0]));
+
+        if (recursive) {
+            shareGroupRunner.apply(new String[]{"find", path, "-type", "d", "-exec", "chmod", "g+s", "{}", "+"});
+        }
+    }
+
     /**
      * Every {@code refs/wip/*} ref (see {@link WipRef}). The committer date is read as a unix
      * count of seconds and converted to millis. {@code %00} (NUL) separates the fields because a

fleetd/src/main/java/dev/ltms/fleet/session/SessionManager.java

@@ -472,6 +472,10 @@ public final class SessionManager implements TurnListener {
         try {
             path = worktrees.add(repoRoot, branch, wt.baseRef());
             worktrees.overlayParity(repoRoot, path, launcher.parityOverlay(preResolvedProfile));
+            // fleetd #185 stage 3: MUST run after overlayParity, not folded into add() — overlayParity
+            // copies more files into the worktree after add() returns, so sharing the group any earlier
+            // leaves those overlay files operator-owned and read-only for a different-uid member.
+            worktrees.shareWithGroup(repoRoot, path);
             handle = launcher.spawn(new SpawnRequest(profile, path, callerCwd, sessionName, resumeSessionId, memberRole));
         } catch (RuntimeException e) {
             log.warn("spawn failed for profile={} role={} branch={} path={}: {}",

fleetd/src/main/java/dev/ltms/fleet/session/Worktrees.java

@@ -98,4 +98,21 @@ public interface Worktrees {
     /** CB-586: the operator-visible census of {@code refs/wip/*} in one repository. */
     record WipRefStats(int count, long costBytes) {
     }
+
+    /**
+     * Make {@code repoRoot}'s git store and {@code worktreePath} writable by the configured group
+     * (fleetd #185 stage 3), so a member spawned as a different OS user (see
+     * {@code memberHerdrSocket}) can write its own worktree, its per-worktree git metadata, and
+     * its own commit objects. No-op when no group is configured.
+     *
+     * <p><strong>This isolates credentials, not the repository.</strong> A member in the group can
+     * still write the operator's git objects and refs in the shared repo — this only fixes file
+     * ownership/permissions so a different-uid member can work at all, it grants no narrower access
+     * than that.
+     *
+     * @param repoRoot     the repository whose git store ({@code .git/objects}, {@code refs},
+     *                     {@code logs}, {@code worktrees}, {@code packed-refs}) needs sharing
+     * @param worktreePath the linked worktree's own directory
+     */
+    void shareWithGroup(String repoRoot, String worktreePath);
 }

fleetd/src/test/java/dev/ltms/fleet/config/FleetConfigTest.java

@@ -1036,6 +1036,35 @@ class FleetConfigTest {
         assertEquals(LeadMailbox.DEFAULT_PREFETCH, noEnv.prefetchOrDefault());
     }
 
+    @Test
+    void absentWorktreeGroupLeavesItNull(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-worktree-group.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\n");
+
+        FleetConfig cfg = FleetConfig.load(f);
+        assertNull(cfg.worktreeGroup(), "no worktreeGroup: key → null → GitWorktrees.shareWithGroup is a no-op");
+    }
+
+    @Test
+    void worktreeGroupKeyParses(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("worktree-group.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\nworktreeGroup: fleet-workers\n");
+
+        FleetConfig cfg = FleetConfig.load(f);
+        assertEquals("fleet-workers", cfg.worktreeGroup());
+    }
+
+    @Test
+    void absentWorktreeGroupSurvivesTheBackCompatConstructorChain() {
+        // fleetd #185 stage 3: withDefaults() (and every pre-existing call site) must not silently
+        // drop a live worktreeGroup by routing through a back-compat constructor that defaults it
+        // to null.
+        FleetConfig cfg = new FleetConfig(null, null, null, Map.of(), null, null, null, null, null,
+                null, null, null, null, null, null, null, null, null, null, null, "fleet-workers");
+        assertEquals("fleet-workers", cfg.withDefaults().worktreeGroup(),
+                "withDefaults() must carry a configured worktreeGroup through unchanged");
+    }
+
     @Test
     void absentPrimaryBlockLeavesPrimaryNull(@TempDir Path dir) throws Exception {
         Path f = dir.resolve("no-primary.yaml");

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

@@ -688,6 +688,32 @@ 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");

fleetd/src/test/java/dev/ltms/fleet/session/FakeWorktrees.java

@@ -30,12 +30,19 @@ public final class FakeWorktrees implements Worktrees {
     public record PruneCall(String repoRoot, long minAgeMillis) {
     }
 
+    public record ShareCall(String repoRoot, String worktreePath) {
+    }
+
     private final List<AddCall> addCalls = new CopyOnWriteArrayList<>();
     private final List<RemoveCall> removeCalls = new CopyOnWriteArrayList<>();
     private final List<OverlayCall> overlayCalls = new CopyOnWriteArrayList<>();
     private final List<RepoRootCall> repoRootCalls = new CopyOnWriteArrayList<>();
     private final List<SnapshotCall> snapshotCalls = new CopyOnWriteArrayList<>();
     private final List<PruneCall> pruneCalls = new CopyOnWriteArrayList<>();
+    private final List<ShareCall> shareCalls = new CopyOnWriteArrayList<>();
+    /** Tags every {@code overlayParity}/{@code shareWithGroup} call in call order, so a test can
+     *  pin that sharing runs after the overlay copy (fleetd #185 stage 3). */
+    private final List<String> overlayShareOrder = new CopyOnWriteArrayList<>();
     private final Set<String> existingPaths = ConcurrentHashMap.newKeySet();
     private final Set<String> trackedPaths = ConcurrentHashMap.newKeySet();
     private final AtomicLong snapshotSeq = new AtomicLong();
@@ -136,6 +143,13 @@ public final class FakeWorktrees implements Worktrees {
         }
         overlayCalls.add(new OverlayCall(repoRoot, worktreePath, List.copyOf(overlay),
                 List.copyOf(copied), List.copyOf(skipped)));
+        overlayShareOrder.add("overlay:" + worktreePath);
+    }
+
+    @Override
+    public void shareWithGroup(String repoRoot, String worktreePath) {
+        shareCalls.add(new ShareCall(repoRoot, worktreePath));
+        overlayShareOrder.add("share:" + worktreePath);
     }
 
     @Override
@@ -203,4 +217,17 @@ public final class FakeWorktrees implements Worktrees {
     public SnapshotCall lastSnapshot() {
         return snapshotCalls.isEmpty() ? null : snapshotCalls.getLast();
     }
+
+    public List<ShareCall> shareCalls() {
+        return List.copyOf(shareCalls);
+    }
+
+    public ShareCall lastShare() {
+        return shareCalls.isEmpty() ? null : shareCalls.getLast();
+    }
+
+    /** Call-order tags ({@code "overlay:<path>"}/{@code "share:<path>"}) — see field javadoc. */
+    public List<String> overlayShareOrder() {
+        return List.copyOf(overlayShareOrder);
+    }
 }

fleetd/src/test/java/dev/ltms/fleet/session/GitWorktreesTest.java

@@ -924,4 +924,179 @@ class GitWorktreesTest {
         assertEquals(2, stats.count(), "two snapshot refs are reported");
         assertTrue(stats.costBytes() > 0, "the cost of the snapshots is a positive byte count");
     }
+
+    /**
+     * fleetd #185 stage 3: a recording {@link java.util.function.Function} test seam stands in for
+     * every process {@link GitWorktrees#shareWithGroup} would run — no real second OS user/group
+     * exists on this host, so these are unit tests against that seam, not a live-group integration
+     * test (out of scope per the ticket).
+     */
+    private static List<String> joined(String[] command) {
+        return List.of(command);
+    }
+
+    /** Remove {@code path} and anything under it. Tolerates an already-absent path. */
+    private static void deleteRecursively(Path path) throws Exception {
+        if (!Files.exists(path)) {
+            return;
+        }
+        if (Files.isDirectory(path)) {
+            try (java.util.stream.Stream<Path> children = Files.list(path)) {
+                for (Path child : children.toList()) {
+                    deleteRecursively(child);
+                }
+            }
+        }
+        Files.delete(path);
+    }
+
+    /** {@code worktreeGroup} absent ⇒ zero processes spawned and no git config written. */
+    @Test
+    void shareWithGroupIsNoopWhenNoGroupConfigured(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        List<List<String>> recorded = new java.util.ArrayList<>();
+        java.util.function.Function<String[], String> recordingRunner = cmd -> {
+            recorded.add(joined(cmd));
+            return "";
+        };
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString(), null, _ -> {}, recordingRunner);
+
+        gitWorktrees.shareWithGroup(repo.toString(), repo.resolve("some-worktree").toString());
+
+        assertTrue(recorded.isEmpty(), "no group configured must spawn no process at all: " + recorded);
+    }
+
+    /** A configured group runs {@code git config core.sharedRepository group} first, then
+     *  chgrp/chmod/setgid over every path {@link GitWorktrees#shareWithGroup} documents. */
+    @Test
+    void shareWithGroupRunsConfigThenChgrpChmodSetgidPerPath(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        String repoRoot = repo.toString();
+        Path worktree = Files.createDirectories(repo.resolve("some-worktree"));
+        String worktreePath = worktree.toString();
+        Files.createDirectories(repo.resolve(".git/worktrees"));
+        List<List<String>> recorded = new java.util.ArrayList<>();
+        java.util.function.Function<String[], String> recordingRunner = cmd -> {
+            recorded.add(joined(cmd));
+            // What real git answers for an ordinary (non-linked) checkout: relative to repoRoot.
+            return List.of(cmd).contains("--git-common-dir") ? ".git\n" : "";
+        };
+        GitWorktrees gitWorktrees =
+                new GitWorktrees(tmp.resolve("wts").toString(), "devteam", _ -> {}, recordingRunner);
+
+        gitWorktrees.shareWithGroup(repoRoot, worktreePath);
+
+        assertEquals(List.of("git", "-C", repoRoot, "config", "core.sharedRepository", "group"), recorded.get(0),
+                "core.sharedRepository must be set first, so it keeps working after the one-time fix-up");
+        assertTrue(recorded.contains(List.of("git", "-C", repoRoot, "rev-parse", "--git-common-dir")),
+                "the git dir must be asked for, never hardcoded as <repoRoot>/.git — that is a FILE "
+                        + "when the checkout is itself a linked worktree: " + recorded);
+
+        for (String dir : List.of(worktreePath, repoRoot + "/.git/objects", repoRoot + "/.git/refs",
+                repoRoot + "/.git/logs", repoRoot + "/.git/worktrees")) {
+            assertTrue(recorded.contains(List.of("chgrp", "-R", "devteam", dir)), "missing chgrp -R for " + dir);
+            assertTrue(recorded.contains(List.of("chmod", "-R", "g+rwX", dir)), "missing chmod -R for " + dir);
+            assertTrue(recorded.contains(List.of("find", dir, "-type", "d", "-exec", "chmod", "g+s", "{}", "+")),
+                    "missing setgid find pass for " + dir);
+        }
+        // packed-refs does not exist in a freshly-init'd repo (only git gc / pack-refs creates it) —
+        // tolerated absence, so it must not appear at all: no recursive/-R treatment for a plain file.
+        String packedRefs = repoRoot + "/.git/packed-refs";
+        assertTrue(recorded.stream().noneMatch(c -> c.contains(packedRefs)),
+                "packed-refs is absent here and must be skipped, not chgrp'd: " + recorded);
+    }
+
+    /**
+     * A path that does not exist is skipped, never handed to {@code chgrp}. {@code .git/logs} is
+     * absent whenever {@code core.logAllRefUpdates} is false or no ref has been updated yet, and
+     * {@code chgrp} on a missing path exits non-zero — which would fail EVERY provisioning spawn
+     * with a message blaming a group that is in fact fine.
+     */
+    @Test
+    void shareWithGroupSkipsPathsThatDoNotExist(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        String repoRoot = repo.toString();
+        deleteRecursively(repo.resolve(".git/logs"));
+        assertFalse(Files.exists(repo.resolve(".git/logs")), "fixture: .git/logs must be gone");
+        List<List<String>> recorded = new java.util.ArrayList<>();
+        java.util.function.Function<String[], String> recordingRunner = cmd -> {
+            recorded.add(joined(cmd));
+            return List.of(cmd).contains("--git-common-dir") ? ".git\n" : "";
+        };
+        GitWorktrees gitWorktrees =
+                new GitWorktrees(tmp.resolve("wts").toString(), "devteam", _ -> {}, recordingRunner);
+
+        gitWorktrees.shareWithGroup(repoRoot, repo.resolve("no-such-worktree").toString());
+
+        String logs = repoRoot + "/.git/logs";
+        assertTrue(recorded.stream().noneMatch(c -> c.contains(logs)),
+                "a missing .git/logs must be skipped, not chgrp'd: " + recorded);
+        assertTrue(recorded.stream().noneMatch(c -> c.contains(repo.resolve("no-such-worktree").toString())),
+                "a missing worktree path must be skipped too: " + recorded);
+        assertTrue(recorded.contains(List.of("chgrp", "-R", "devteam", repoRoot + "/.git/objects")),
+                "paths that DO exist are still shared: " + recorded);
+    }
+
+    /**
+     * The git store is located by {@code rev-parse --git-common-dir}, not by appending
+     * {@code /.git}. When git answers with an absolute path — what it does for a linked worktree,
+     * where {@code <repoRoot>/.git} is a file — every shared path must follow that answer.
+     */
+    @Test
+    void shareWithGroupFollowsAnAbsoluteGitCommonDir(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        Path realGitDir = repo.resolve(".git");
+        List<List<String>> recorded = new java.util.ArrayList<>();
+        java.util.function.Function<String[], String> recordingRunner = cmd -> {
+            recorded.add(joined(cmd));
+            return List.of(cmd).contains("--git-common-dir") ? realGitDir + "\n" : "";
+        };
+        GitWorktrees gitWorktrees =
+                new GitWorktrees(tmp.resolve("wts").toString(), "devteam", _ -> {}, recordingRunner);
+
+        gitWorktrees.shareWithGroup(tmp.resolve("some/linked/worktree").toString(),
+                repo.resolve("wt").toString());
+
+        assertTrue(recorded.contains(List.of("chgrp", "-R", "devteam", realGitDir + "/objects")),
+                "objects must be taken from the reported common dir, not <repoRoot>/.git: " + recorded);
+    }
+
+    /** {@code packed-refs}, when present, is chgrp/chmod'd but never setgid'd (it is a file, not a dir). */
+    @Test
+    void shareWithGroupIncludesPackedRefsWhenPresent(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        String repoRoot = repo.toString();
+        Path packedRefsPath = repo.resolve(".git/packed-refs");
+        Files.writeString(packedRefsPath, "");
+        List<List<String>> recorded = new java.util.ArrayList<>();
+        java.util.function.Function<String[], String> recordingRunner = cmd -> {
+            recorded.add(joined(cmd));
+            return "";
+        };
+        GitWorktrees gitWorktrees =
+                new GitWorktrees(tmp.resolve("wts").toString(), "devteam", _ -> {}, recordingRunner);
+
+        gitWorktrees.shareWithGroup(repoRoot, repo.resolve("some-worktree").toString());
+
+        String packedRefs = packedRefsPath.toString();
+        assertTrue(recorded.contains(List.of("chgrp", "devteam", packedRefs)),
+                "packed-refs must be chgrp'd non-recursively when present: " + recorded);
+        assertTrue(recorded.contains(List.of("chmod", "g+rwX", packedRefs)),
+                "packed-refs must be chmod'd non-recursively when present: " + recorded);
+        assertTrue(recorded.stream().noneMatch(c -> c.contains("find") && c.contains(packedRefs)),
+                "packed-refs (a file) must never get the recursive setgid pass: " + recorded);
+    }
+
+    /** A group that does not exist (or that the operator is not a member of) fails loudly, naming it. */
+    @Test
+    void shareWithGroupThrowsNamingTheGroupWhenChgrpFails(@TempDir Path tmp) throws Exception {
+        Path repo = initRepo(tmp.resolve("repo"));
+        GitWorktrees gitWorktrees = new GitWorktrees(tmp.resolve("wts").toString(), "cb185-nonexistent-group-zz");
+        String wt = new GitWorktrees(tmp.resolve("wts").toString()).add(repo.toString(), "cb-185-share", "HEAD");
+
+        WorktreeException e = assertThrows(WorktreeException.class,
+                () -> gitWorktrees.shareWithGroup(repo.toString(), wt));
+        assertTrue(e.getMessage().contains("cb185-nonexistent-group-zz"),
+                "exception must name the missing/refused group: " + e.getMessage());
+    }
 }

fleetd/src/test/java/dev/ltms/fleet/session/SessionManagerTest.java

@@ -148,6 +148,10 @@ class SessionManagerTest {
             return 0;
         }
 
+        @Override
+        public void shareWithGroup(String repoRoot, String worktreePath) {
+        }
+
         List<String> removeCalls() {
             return List.copyOf(removeCalls);
         }

fleetd/src/test/java/dev/ltms/fleet/session/WorktreeSessionManagerTest.java

@@ -163,6 +163,37 @@ class WorktreeSessionManagerTest {
                 "tracked copied paths are --skip-worktree'd");
     }
 
+    /**
+     * fleetd #185 stage 3, THE TRAP: {@code overlayParity} copies more files into the worktree
+     * AFTER {@code add} returns, so {@code shareWithGroup} must run after it, not folded into
+     * {@code add()} — otherwise every overlay file lands operator-owned and unwritable for a
+     * different-uid member, with a green test suite hiding it.
+     */
+    @Test
+    void shareWithGroupRunsAfterOverlayParityNotBeforeIt() {
+        FakeHerdr herdr = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
+                .track(".envrc");
+        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
+
+        MemberSession s = sessions.acquire("ltms-local", null, "/caller/proj", null,
+                new WorktreeRequest("cb-185", null));
+
+        assertEquals(1, worktrees.overlayCalls().size(), "overlayParity ran exactly once");
+        assertEquals(1, worktrees.shareCalls().size(), "shareWithGroup ran exactly once");
+        FakeWorktrees.OverlayCall overlay = worktrees.lastOverlay();
+        FakeWorktrees.ShareCall share = worktrees.lastShare();
+        assertEquals(s.worktree(), overlay.worktreePath());
+        assertEquals(s.worktree(), share.worktreePath());
+
+        List<String> order = worktrees.overlayShareOrder();
+        int overlayIndex = order.indexOf("overlay:" + s.worktree());
+        int shareIndex = order.indexOf("share:" + s.worktree());
+        assertTrue(overlayIndex >= 0 && shareIndex >= 0, "both calls must be recorded: " + order);
+        assertTrue(overlayIndex < shareIndex,
+                "shareWithGroup MUST run after overlayParity, not before/inside add(): " + order);
+    }
+
     @Test
     void releaseRemovesWorktreeButDoesNotDeleteBranch() {
         FakeHerdr herdr = new FakeHerdr();