#137: complete an async ticket's own reply after answer() times out

fleet_send{turnId} (MessageService.answer) blocks the primary only for its
own bounded MCP-call window (25s default, 120s max) — far shorter than a
worker's resumed turn can genuinely take. When that window expires, answer()
closes its rendezvous waiter, so the worker's eventual fleet_reply has no
live waiter to resolve and falls back to the session inbox. The async
ticket's future was never completed by that path, so fleet_poll{ticket}
stayed PENDING until fleet_stop's abandon() forced it FAILED with a
misleading "the worker session was released before it replied" reason,
even though the reply had genuinely arrived.

- MessageService.reply(): before falling to the inbox, look for the async
  task this exact turn belongs to (already answered — question cleared,
  turnId still stamped — but not yet resolved) and complete it directly
  with the real reply, so fleet_poll{ticket} returns it.
- MessageService.abandon(): defense in depth, independent of the above —
  never write a false WORKER_FAILED once a reply reached the inbox for
  this target; recover and use its real content instead.
- Two new tests drive the full delegation path (async send -> ask ->
  answer with a short timeout -> reply -> poll/abandon), not a reply sink
  directly; both fail with the fix disabled and pass with it restored.

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)

REPORT-cb175.md

@@ -1,52 +0,0 @@
-# CB-175 report
-
-## Change
-
-`OpenCodeLauncher` reads the newest opencode session record for the worker cwd after spawn readiness.
-It compares the requested `provider/model` selector with `model.providerID/model.id` from the record.
-`variant` is not compared because a profile selector has no variant part.
-
-An absent record, malformed record, or incomplete model object is unknown evidence. It does not log
-an error or quarantine the profile.
-
-On a real mismatch, fleetd logs an ERROR with the requested and resolved selectors. The mismatch goes
-through `ExhaustionSink` into the existing `BackendQuarantine` and uses the profile's
-`effectiveCredentialId()`.
-
-I chose a permanent, process-lifetime quarantine. A withdrawn selector cannot become correct after a
-cooldown. A timed retry could silently use the paid fallback again. `fleet_list` will show the usual
-quarantine state, with a very large remaining time, until fleetd restarts after an operator fixes the
-profile.
-
-## Tests
-
-Added tests for an exact match, a mismatch, and missing or unreadable session storage.
-
-I proved the mismatch test fails without the quarantine call. I commented out the call and ran:
-
-```text
-mvn -Dtest=OpenCodeLauncherTest#differentResolvedModelPermanentlyQuarantinesTheProfile test
-```
-
-The result was:
-
-```text
-[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0
-org.opentest4j.AssertionFailedError: a fallback model must block later spawns ==> expected: <true> but was: <false>
-[INFO] BUILD FAILURE
-```
-
-I restored the call. I then ran `mvn clean install` in `fleetd/` without a pipe. Its result was:
-
-```text
-[INFO] Tests run: 1040, Failures: 0, Errors: 0, Skipped: 0
-[INFO] BUILD SUCCESS
-```
-
-## Limits and scope
-
-I could not spawn a real opencode member or restart fleetd. I did not test this end to end against a
-live opencode session database.
-
-I confirmed `ClaudeCodeLauncher` passes `--model` but does not read back the resolved model. I did
-not change it because it is outside this ticket's scope.

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

@@ -168,17 +168,6 @@ public final class Fleetd {
                 claudeProfiles.put(name, w);
             }
         });
-        // One tracker covers timed backend exhaustion and permanent model-selector mismatches. The
-        // latter cannot heal on a retry, so OpenCodeLauncher uses quarantinePermanently through the
-        // sink below rather than letting a cooldown reopen a paid fallback.
-        BackendQuarantine quarantine = new BackendQuarantine(System::nanoTime,
-                TimeUnit.SECONDS.toNanos(cfg.quarantineCooldownSeconds()));
-        ExhaustionSink modelMismatchSink = (profileName, reason) -> {
-            FleetConfig.Profile profile = config.get().profiles().get(profileName);
-            if (profile != null) {
-                quarantine.quarantinePermanently(profile.effectiveCredentialId());
-            }
-        };
         List<HerdrPeerLauncher> adapters = new ArrayList<>();
         // The claude-code adapter is the always-present default; keep it even with no profiles (so a
         // bridge configured with no workers, or opencode-only, still has a well-defined base adapter)
@@ -195,13 +184,15 @@ public final class Fleetd {
                     opencodeProfiles, cfg.effectiveDefaultProfile(), System::getenv,
                     cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs(),
                     () -> config.get().fleet(),
-                    () -> config.get().memberCredentials(), modelMismatchSink));
+                    () -> config.get().memberCredentials()));
         }
         AtomicReference<Function<String, Integer>> liveCountRef = new AtomicReference<>(_ -> 0);
         // CB-578 stage B: one quarantine tracker for the whole daemon, shared between the launcher
         // (checked at spawn) and the exhaustion sink wired in below (written on BACKEND_EXHAUSTED).
         // The cooldown is deferred (see FleetConfig#quarantineCooldownSeconds): it is read once
         // here, at startup, and a config reload only changes it for a daemon restart.
+        BackendQuarantine quarantine = new BackendQuarantine(System::nanoTime,
+                TimeUnit.SECONDS.toNanos(cfg.quarantineCooldownSeconds()));
         PeerLauncher workers = new CompositePeerLauncher(
                 adapters,
                 cfg.effectiveDefaultProfile(),

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

@@ -6,7 +6,6 @@ import dev.ltms.fleet.config.FleetConfig;
 import dev.ltms.fleet.herdr.Agent;
 import dev.ltms.fleet.herdr.AgentControl;
 import dev.ltms.fleet.herdr.WorkspaceControl;
-import dev.ltms.fleet.inject.ExhaustionSink;
 import dev.ltms.fleet.peer.Capability;
 import dev.ltms.fleet.peer.CharterReceipt;
 import dev.ltms.fleet.peer.PeerHandle;
@@ -72,9 +71,6 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
      */
     private final OpenCodeSessionDiscovery discovery;
 
-    /** Receives the profile when a resolved model differs from its requested selector. */
-    private final ExhaustionSink modelMismatchSink;
-
     /**
      * Production constructor — disables the spawn-ready gate ({@code spawnReadyTimeoutMs == 0}) so it
      * matches the legacy non-blocking spawn semantics. Config dirs are created under the JVM temp dir.
@@ -125,20 +121,7 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
                             Supplier<FleetConfig.MemberCredentials> memberCredentials) {
         this(agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs,
                 System::currentTimeMillis, () -> sleepUninterruptibly(spawnReadyPollMs),
-                defaultConfigRoot(), defaultDiscoveryRoot(), fleet, memberCredentials, ExhaustionSink.none());
-    }
-
-    /** Production constructor with permanent-quarantine wiring for a model mismatch. */
-    public OpenCodeLauncher(AgentControl agents, WorkspaceControl spaces,
-                             Map<String, FleetConfig.Profile> profiles, String defaultProfile,
-                             Function<String, String> env,
-                             long spawnReadyTimeoutMs, long spawnReadyPollMs,
-                             Supplier<FleetConfig.Fleet> fleet,
-                             Supplier<FleetConfig.MemberCredentials> memberCredentials,
-                             ExhaustionSink modelMismatchSink) {
-        this(agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs,
-                System::currentTimeMillis, () -> sleepUninterruptibly(spawnReadyPollMs),
-                defaultConfigRoot(), defaultDiscoveryRoot(), fleet, memberCredentials, modelMismatchSink);
+                defaultConfigRoot(), defaultDiscoveryRoot(), fleet, memberCredentials);
     }
 
     /**
@@ -180,10 +163,12 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
                             Function<String, String> env,
                             long spawnReadyTimeoutMs,
                             LongSupplier nowMillis, Runnable sleeper,
-                             Path configRoot, Path discoveryRoot,
-                             Supplier<FleetConfig.Fleet> fleet) {
-        this(agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs, nowMillis, sleeper,
-                configRoot, discoveryRoot, fleet, null, ExhaustionSink.none());
+                            Path configRoot, Path discoveryRoot,
+                            Supplier<FleetConfig.Fleet> fleet) {
+        super(NAME_PREFIX, agents, spaces, profiles, defaultProfile, env,
+                spawnReadyTimeoutMs, nowMillis, sleeper, fleet);
+        this.configRoot = configRoot;
+        this.discovery = new OpenCodeSessionDiscovery(discoveryRoot);
     }
 
     /**
@@ -194,32 +179,13 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
                             Function<String, String> env,
                             long spawnReadyTimeoutMs,
                             LongSupplier nowMillis, Runnable sleeper,
-                             Path configRoot, Path discoveryRoot,
-                             Supplier<FleetConfig.Fleet> fleet,
-                             Supplier<FleetConfig.MemberCredentials> memberCredentials) {
-        this(agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs, nowMillis, sleeper,
-                configRoot, discoveryRoot, fleet, memberCredentials, ExhaustionSink.none());
-    }
-
-    /**
-     * Full constructor with the model-mismatch quarantine callback. The callback is an
-     * {@link ExhaustionSink} so model mismatches use the existing quarantine path rather than a
-     * second state tracker.
-     */
-    public OpenCodeLauncher(AgentControl agents, WorkspaceControl spaces,
-                             Map<String, FleetConfig.Profile> profiles, String defaultProfile,
-                             Function<String, String> env,
-                             long spawnReadyTimeoutMs,
-                             LongSupplier nowMillis, Runnable sleeper,
-                             Path configRoot, Path discoveryRoot,
-                             Supplier<FleetConfig.Fleet> fleet,
-                             Supplier<FleetConfig.MemberCredentials> memberCredentials,
-                             ExhaustionSink modelMismatchSink) {
+                            Path configRoot, Path discoveryRoot,
+                            Supplier<FleetConfig.Fleet> fleet,
+                            Supplier<FleetConfig.MemberCredentials> memberCredentials) {
         super(NAME_PREFIX, agents, spaces, profiles, defaultProfile, env,
                 spawnReadyTimeoutMs, nowMillis, sleeper, fleet, memberCredentials);
         this.configRoot = configRoot;
         this.discovery = new OpenCodeSessionDiscovery(discoveryRoot);
-        this.modelMismatchSink = modelMismatchSink;
     }
 
     private static Path defaultConfigRoot() {
@@ -532,34 +498,9 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
     @Override
     public PeerHandle spawn(SpawnRequest req) {
         PeerHandle inner = super.spawn(req);
-        verifyResolvedModel(requireProfile(req.profileName()), effectiveCwd(req));
         return new SessionAwareHandle(inner, discovery, effectiveCwd(req));
     }
 
-    /**
-     * Read the session record once after the spawn-readiness gate. opencode writes the model it
-     * actually selected there. No record, incomplete model object, or a selector without one slash
-     * is unknown evidence, so it must not quarantine a working profile.
-     */
-    private void verifyResolvedModel(FleetConfig.Profile cfg, String cwd) {
-        String[] requested = splitProviderModel(cfg.model());
-        if (requested == null) {
-            return;
-        }
-        OpenCodeSessionDiscovery.SessionRecord record = discovery.sessionForDirectory(cwd);
-        if (record == null || record.providerId() == null || record.modelId() == null) {
-            return;
-        }
-        String actual = record.providerId() + "/" + record.modelId();
-        if (cfg.model().equals(actual)) {
-            return;
-        }
-        log.error("opencode model mismatch for profile '{}': requested '{}' but resolved '{}'",
-                cfg.profile(), cfg.model(), actual);
-        modelMismatchSink.onExhausted(cfg.profile(), "opencode model mismatch: requested "
-                + cfg.model() + ", resolved " + actual);
-    }
-
     /**
      * A {@link PeerHandle} that delegates everything to the base's worker handle but resolves
      * {@link #agentSessionId()} lazily through opencode session discovery. Delegate-only, so the

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

@@ -58,15 +58,6 @@ final class OpenCodeSessionDiscovery {
      * @return the matching session id, or {@code null} if none is known yet
      */
     String sessionIdForDirectory(String directory) {
-        SessionRecord record = sessionForDirectory(directory);
-        return record == null ? null : record.id();
-    }
-
-    /**
-     * The newest session record for {@code directory}, or {@code null} when opencode has not written
-     * one yet. This is the single storage seam for both session identity and resolved-model checks.
-     */
-    SessionRecord sessionForDirectory(String directory) {
         if (directory == null || directory.isBlank()) {
             return null;
         }
@@ -74,20 +65,20 @@ final class OpenCodeSessionDiscovery {
         if (!Files.isDirectory(sessionRoot)) {
             return null;
         }
-        SessionRecord best = null;
+        String best = null;
         long bestMtime = Long.MIN_VALUE;
         try (Stream<Path> projectDirs = Files.list(sessionRoot)) {
             for (Path projectDir : projectDirs.filter(Files::isDirectory).toList()) {
                 try (Stream<Path> records = Files.list(projectDir)) {
                     for (Path record : records.toList()) {
-                        SessionRecord matched = matchRecord(record, directory);
-                        if (matched == null) {
+                        String id = matchId(record, directory);
+                        if (id == null) {
                             continue;
                         }
                         long mtime = lastModifiedEpochMillis(record);
                         if (mtime > bestMtime) {
                             bestMtime = mtime;
-                            best = matched;
+                            best = id;
                         }
                     }
                 } catch (IOException ignored) {
@@ -106,7 +97,7 @@ final class OpenCodeSessionDiscovery {
      * that is not JSON, lacks {@code id}/{@code directory}, or points at a different directory is
      * simply not our session; a malformed one is skipped, never fatal.
      */
-    private SessionRecord matchRecord(Path record, String directory) {
+    private String matchId(Path record, String directory) {
         try {
             JsonNode node = json.readTree(record.toFile());
             JsonNode id = node == null ? null : node.get("id");
@@ -114,24 +105,12 @@ final class OpenCodeSessionDiscovery {
             if (id == null || dir == null || !directory.equals(dir.asText())) {
                 return null;
             }
-            JsonNode model = node.path("model");
-            String providerId = text(model, "providerID");
-            String modelId = text(model, "id");
-            return new SessionRecord(id.asText(), providerId, modelId);
+            return id.asText();
         } catch (IOException e) {
             return null;
         }
     }
 
-    private static String text(JsonNode node, String name) {
-        JsonNode value = node.get(name);
-        return value == null || value.isNull() || value.asText().isBlank() ? null : value.asText();
-    }
-
-    /** The opencode fields fleetd reads from one session record. Null model fields mean unknown. */
-    record SessionRecord(String id, String providerId, String modelId) {
-    }
-
     /** The record's last-modified epoch ms, or {@code Long.MIN_VALUE} if unreadable (never wins). */
     private static long lastModifiedEpochMillis(Path record) {
         try {

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

@@ -366,21 +366,40 @@ public final class MessageService {
     }
 
     /**
-     * Route a worker's explicit {@code fleet_reply}: resolve an open send, or queue it in the
-     * inbox if no send is currently open. Unlike the bare {@link Rendezvous#resolve}, a no-waiter
-     * result is <em>not</em> a failure — the reply is held for later drain.
+     * Route a worker's explicit {@code fleet_reply}: resolve an open send, complete an async ticket
+     * still parked waiting on this exact turn's answer, or — only once neither applies — queue it in
+     * the inbox. Unlike the bare {@link Rendezvous#resolve}, a no-waiter result is <em>not</em> a
+     * failure — the reply is held for later drain.
      *
      * <p><strong>Do NOT use this for mid-turn questions.</strong> {@code fleet_ask} /
      * {@link Rendezvous#resolveQuestion} must keep today's {@code NO_WAITER} behaviour — questions
      * are interactive and must never be queued.
      *
-     * @return always {@code true} — the reply either resolved a live send or was queued
+     * @return always {@code true} — the reply resolved a live send, completed a parked ticket, or
+     *         was queued
      */
     public boolean reply(String session, String content) {
         if (rendezvous.resolve(session, content)) {
             count(FleetMetrics.REPLIES, "path", "rendezvous");
             return true;                       // a live send took it — unchanged fast path
         }
+        // #137: no live rendezvous waiter, but this may be the worker's real fleet_reply resuming a
+        // turn that {@link #answer} already gave up waiting on. answer()'s own bounded wait (the
+        // primary's fleet_send{turnId} call, capped well under a minute) can time out and close its
+        // waiter long before the worker — now actually resuming real work — finishes and replies. That
+        // reply used to have nowhere to land but the session inbox, leaving the async ticket's future
+        // unresolved forever: fleet_poll{ticket} stayed PENDING until fleet_stop's abandon() forced it
+        // FAILED with a misleading "session released before it replied" reason, even though the reply
+        // had, in fact, arrived. Completing the matching ticket directly here means fleet_poll{ticket}
+        // sees the real reply instead.
+        Task orphan = askAnsweredAsyncTask(session);
+        if (orphan != null && 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
+        }
         inbox.publish(session, UUID.randomUUID().toString(), content);
         // CB-640: record the stranding itself (not just the reply text) so fleet health can see a
         // worker whose replies keep missing their waiter, not only the queue depth this leaves behind.
@@ -394,6 +413,24 @@ public final class MessageService {
         return true;                           // held, not lost
     }
 
+    /**
+     * 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}).
+     */
+    private Task askAnsweredAsyncTask(String target) {
+        for (Task task : tasks.values()) {
+            if (target.equals(task.target) && task.question == null && task.turnId != null
+                    && !task.future.isDone()) {
+                return task;
+            }
+        }
+        return null;
+    }
+
     /** Record a counter sample when a registry is wired; a no-op in unit tests. */
     private void count(String name, String... labels) {
         if (metrics != null) {
@@ -435,19 +472,43 @@ public final class MessageService {
      * <p>Resolving the waiter as a failure — rather than letting it time out — also means the
      * outcome is counted, so a torn-down delegation stops being invisible to {@code /metrics}.
      *
-     * @return true if a live waiter was failed
+     * <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
+     * {@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).
+     *
+     * @return true if a live waiter or an async task was failed (never true for one recovered as a
+     *         reply — see the note above)
      */
     public boolean abandon(String target, String reason) {
+        boolean hadStrandedReply = hasStrandedReply(target);
         // CB-640: the session is gone — nothing will ever accept or deliver into it now.
         strandedReplies.remove(target);
         queuedDeliveries.remove(target);
         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
         for (Task task : tasks.values()) {
-            if (target.equals(task.target) && task.question == null
-                    && task.future.complete(new Reply(Outcome.WORKER_FAILED, reason))) {
-                asyncFailed = true;
+            if (!target.equals(task.target) || task.question != null || task.future.isDone()) {
+                continue;
+            }
+            if (hadStrandedReply && recovered == null) {
+                recovered = recoverStrandedReply(target);
+            }
+            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);
+                }
             }
         }
         if (failed) {
@@ -456,6 +517,21 @@ public final class MessageService {
         return failed || asyncFailed;
     }
 
+    /**
+     * Drain {@code target}'s inbox and hand its content back as a {@link Outcome#REPLIED} result
+     * (#137 defence in depth for {@link #abandon}) — {@code null} if it turned out empty (the
+     * 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).
+     */
+    private Reply recoverStrandedReply(String target) {
+        var messages = drainReplies(target);
+        if (messages.isEmpty()) {
+            return null;
+        }
+        return new Reply(Outcome.REPLIED, messages.get(messages.size() - 1).content());
+    }
+
     /**
      * Acknowledge a specific reply by {@code msgId} for {@code target}. Removes it from the inbox
      * so that a subsequent drain or peek no longer returns it.

fleetd/src/main/java/dev/ltms/fleet/placement/BackendQuarantine.java

@@ -76,20 +76,6 @@ public final class BackendQuarantine {
         quarantinedUntilNanos.put(credentialId, nowNanos.getAsLong() + cooldownNanos);
     }
 
-    /**
-     * Quarantine {@code credentialId} until the daemon restarts. This is for a configuration error
-     * that cannot heal with time, unlike an exhausted backend. A model selector that opencode silently
-     * resolves to another model stays wrong until an operator changes the profile, so a cooldown would
-     * start the same unsafe work again.
-     */
-    public void quarantinePermanently(String credentialId) {
-        Objects.requireNonNull(credentialId, "credentialId");
-        if (inert) {
-            return;
-        }
-        quarantinedUntilNanos.put(credentialId, Long.MAX_VALUE);
-    }
-
     /** Whether {@code credentialId} is quarantined right now. */
     public boolean isQuarantined(String credentialId) {
         return remainingNanos(credentialId) > 0;
@@ -124,7 +110,6 @@ public final class BackendQuarantine {
     }
 
     private static long toSecondsRoundedUp(long nanos) {
-        long seconds = nanos / 1_000_000_000L;
-        return seconds + (nanos % 1_000_000_000L == 0 ? 0 : 1);
+        return (nanos + 999_999_999L) / 1_000_000_000L;
     }
 }

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

@@ -12,7 +12,6 @@ import dev.ltms.fleet.peer.MemberRole;
 import dev.ltms.fleet.peer.PeerHandle;
 import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.peer.SpawnRequest;
-import dev.ltms.fleet.placement.BackendQuarantine;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -23,7 +22,6 @@ import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -63,23 +61,6 @@ class OpenCodeLauncherTest {
                 0, System::currentTimeMillis, () -> { }, configRoot, configRoot, null, () -> creds);
     }
 
-    private static OpenCodeLauncher serviceWithModelMismatchSink(FakeHerdr herdr, Path root,
-                                                                  FleetConfig.Profile cfg,
-                                                                  BackendQuarantine quarantine) {
-        return new OpenCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
-                Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null,
-                0, System::currentTimeMillis, () -> { }, root, root, null, null,
-                (profile, _) -> quarantine.quarantinePermanently(profile));
-    }
-
-    private static void writeResolvedModelRecord(Path root, String directory, String providerId,
-                                                 String modelId) throws Exception {
-        Path record = Files.createDirectories(root.resolve("session").resolve("p1")).resolve("ses_a.json");
-        Files.writeString(record, "{\"id\":\"ses_a\",\"directory\":\"" + directory
-                + "\",\"model\":{\"id\":\"" + modelId + "\",\"providerID\":\""
-                + providerId + "\",\"variant\":\"high\"}}");
-    }
-
     @SuppressWarnings("unchecked")
     private static Map<String, Object> lastStart(FakeHerdr herdr) {
         return (Map<String, Object>) herdr.lastCall("agent.start").params();
@@ -230,50 +211,6 @@ class OpenCodeLauncherTest {
                 "--auto is unconditional: a model-less worker still must never block on approval");
     }
 
-    // --- CB-175: verify opencode's recorded resolved model after spawn readiness ----------------
-
-    @Test
-    void matchingResolvedModelDoesNotQuarantineTheProfile(@TempDir Path root) throws Exception {
-        FleetConfig.Profile cfg = opencodeCfg("opencode/x-preview-f-free", null, null);
-        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(1));
-        writeResolvedModelRecord(root, root.toString(), "opencode", "x-preview-f-free");
-
-        serviceWithModelMismatchSink(new FakeHerdr(), root, cfg, quarantine)
-                .spawn(new SpawnRequest(null, root.toString(), null, null, null, MemberRole.DEV));
-
-        assertFalse(quarantine.isQuarantined("gemini"), "the exact provider/model match is safe");
-    }
-
-    @Test
-    void differentResolvedModelPermanentlyQuarantinesTheProfile(@TempDir Path root) throws Exception {
-        FleetConfig.Profile cfg = opencodeCfg("opencode/x-preview-f-free", null, null);
-        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(1));
-        writeResolvedModelRecord(root, root.toString(), "openai", "gpt-5.6-sol");
-
-        serviceWithModelMismatchSink(new FakeHerdr(), root, cfg, quarantine)
-                .spawn(new SpawnRequest(null, root.toString(), null, null, null, MemberRole.DEV));
-
-        assertTrue(quarantine.isQuarantined("gemini"), "a fallback model must block later spawns");
-        assertEquals(Long.MAX_VALUE / 1_000_000_000L + 1, quarantine.remainingSeconds("gemini").orElseThrow(),
-                "a withdrawn selector cannot become safe after the normal cooldown");
-    }
-
-    @Test
-    void missingOrUnreadableSessionDatabaseDoesNotQuarantine(@TempDir Path root) throws Exception {
-        FleetConfig.Profile cfg = opencodeCfg("opencode/x-preview-f-free", null, null);
-        BackendQuarantine missing = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(1));
-        serviceWithModelMismatchSink(new FakeHerdr(), root, cfg, missing)
-                .spawn(new SpawnRequest(null, root.toString(), null, null, null, MemberRole.DEV));
-        assertFalse(missing.isQuarantined("gemini"), "a missing database is unknown evidence");
-
-        Path broken = Files.createDirectories(root.resolve("session").resolve("p1")).resolve("ses_a.json");
-        Files.writeString(broken, "not JSON");
-        BackendQuarantine unreadable = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(1));
-        serviceWithModelMismatchSink(new FakeHerdr(), root, cfg, unreadable)
-                .spawn(new SpawnRequest(null, root.toString(), null, null, null, MemberRole.DEV));
-        assertFalse(unreadable.isQuarantined("gemini"), "an unreadable database is unknown evidence");
-    }
-
     // --- CB-617: --agent <role> when the role has an agent-definition file --------------------
 
     @Test

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

@@ -710,6 +710,68 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
     }
 
+    // --- #137: a fleet_ask round-trip must not orphan the ticket's own reply -------------------
+    //
+    // The primary's fleet_send{turnId} answer call is itself bounded (a real MCP call, capped well
+    // under a minute) — far shorter than a resumed turn can genuinely take to finish real work. These
+    // drive the exact real delegation path (async send -> worker asks -> primary answers -> primary's
+    // own wait gives up -> worker's real fleet_reply arrives afterwards) rather than calling a reply
+    // sink directly, since the bug is specifically about which sink the resumed turn's reply reaches.
+
+    @Test
+    void aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+
+        // The primary answers, but its own bounded wait for the worker's resumed turn is short and
+        // expires before the worker (still genuinely working) gets back to it.
+        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150);
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
+                "the primary's own bounded wait gives up before the worker finishes resuming");
+
+        // The worker keeps working past that window and only now calls fleet_reply.
+        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+        MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+        assertEquals("PR opened: https://example/pulls/42", done.reply(),
+                "fleet_poll{ticket} must return the worker's real reply, not stay pending forever");
+        assertEquals("reply", done.replySource());
+        assertFalse(messages.hasStrandedReply(T),
+                "the reply completed its own ticket directly and never touched the inbox");
+    }
+
+    @Test
+    void fleetStopAfterAnOrphanedReplyDoesNotFailTheTicket() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
+
+        MessageService.Reply answerReply = messages.answer(asking.turnId(), "config.yaml", 150);
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome());
+
+        assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
+
+        // fleet_stop tears the worker's session down right after the reply landed — this must never
+        // report the misleading "the worker session was released before it replied": a reply is
+        // exactly what happened.
+        assertFalse(messages.abandon(T, "the worker session was released before it replied"),
+                "a reply already arrived, so nothing here is a genuine failure");
+
+        MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.DONE);
+        assertEquals("PR opened: https://example/pulls/42", view.reply());
+    }
+
     @Test
     void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
         String ticket = messages.sendAsync(T, "task that asks");