fleetd#335: abandon()'s per-task cleanup and sendAsync's terminal hook must not swallow throws

Site 1 (abandon()'s matching loop, reachable): the recovery/put-back branch calls
inbox.publish, which AmqpReplyInbox implements as a real broker round trip that
throws IllegalStateException on an unroutable/unconfirmed/interrupted publish.
An uncaught throw there aborted the loop, stranding every task after it in
`matching` PENDING forever. Fixed by recording each task's own future.complete()
result before any cleanup runs, then wrapping the cleanup in try/catch so one
task's failure cannot stop its siblings from getting their outcome. Reaching the
throwing branch by real timing needs a race the file's own #137 follow-up already
found unreachable through the public API, so the reproducing test uses a
test-only hook (same technique as the existing fleetd #324/#329 hooks) to inject
the throw at that exact point.

Site 2 (sendAsync's task.future.whenComplete, reachable): the returned stage is
discarded, so an uncaught throw from pushLoop.onTicketTerminal vanished with no
log line. Reproduced for real: Fleetd's shutdown hook runs messages.close()
(stops the async executor from taking new work, but does not cancel a send
already in flight) before pushLoop.close() (shuts its scheduler down
immediately) — a ticket completing in that window makes onTicketTerminal's own
scheduler.schedule(...) throw a genuine RejectedExecutionException. Fixed with a
try/catch(Throwable) plus log.error inside the whenComplete action.

Site 3 (the two `finally { asyncTasksByWaiter.remove(reply); rendezvous.close(...);
}` blocks in send() and answer()): read Rendezvous.close/closeAsk and the
ConcurrentHashMap operations behind them — both are plain map ops on a non-null
key with no user-overridable code, so neither can throw. Left unchanged; not a
defect.

Mutation-proven: reverting either fix reproduces the failure it exists to catch
— removing site 1's try/catch aborts abandon() with the injected exception
(MessageServiceTest#aPerTaskCleanupFailureDoesNotStrandTheRemainingMatchingTasks
errors); removing site 2's try/catch leaves the RejectedExecutionException
unlogged (MessageServiceTest#aTicketTerminalPushFailureDoesNotVanishSilently
fails its log assertion). Full suite: mvn clean install, Tests run: 1365,
Failures: 0, Errors: 0, BUILD SUCCESS.

fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java

@@ -380,9 +380,7 @@ public final class CompletionResolver implements TurnListener {
                         + "matched the profile's exhausted pattern): {}", target, reason);
                 // CB-578 stage B: only on the resolution that actually won the race — a late
                 // duplicate must never quarantine a credential twice for one refusal.
-                if (startsWithExhaustion(matchedLine, exhausted)) {
-                    exhaustionSink.onExhausted(target, reason);
-                }
+                exhaustionSink.onExhausted(target, reason);
             }
             return;
         }
@@ -480,9 +478,7 @@ public final class CompletionResolver implements TurnListener {
                         + "usable assistant block; no fleet_reply): {}", target, reason);
                 // CB-578 stage B: only on the resolution that actually won the race — a late
                 // duplicate must never quarantine a credential twice for one refusal.
-                if (startsWithExhaustion(matchedLine, exhausted)) {
-                    exhaustionSink.onExhausted(target, reason);
-                }
+                exhaustionSink.onExhausted(target, reason);
             }
             return true;
         }
@@ -618,38 +614,19 @@ public final class CompletionResolver implements TurnListener {
     }
 
     /**
-     * True when the pattern appears before the first sentence ending in the matched pane line, rather
-     * than after a member has started prose about it.
+     * True when the error pattern begins the matched pane line, rather than appearing in prose.
      *
      * <p>Leading terminal chrome is skipped first — box-drawing characters, bullets, gutter bars and
-     * spaces. Exhaustion patterns often name only the decisive words in a provider message, such as
-     * {@code "usage limit has been reached"}; they do not include its leading {@code "The"}. A bare
-     * {@code lookingAt} would therefore reject that genuine refusal, including one behind terminal
-     * chrome.
+     * spaces. #339 introduced this check with a bare {@code lookingAt}, and that rejected a genuine
+     * error line rendered as {@code "| 503 Service Unavailable: ..."}: the send still failed, but the
+     * credential outage was never recorded. That is the false negative #339's own invariant 3 called
+     * worse than the false positive it set out to fix — measured with a throwaway probe on the
+     * raw-scrape path, which is exactly the path whose comment says to expect leading chrome.
      *
-     * <p>A member's prose about a refusal normally follows a completed sentence. That sentence ending
-     * is enough to keep it from reaching a cooldown sink while the send still fails with the full pane
-     * tail. This is deliberately less strict than the backend-error check because exhausted patterns
-     * may start after a provider's leading words.
+     * <p>Skipping only a leading run of non-letter, non-digit characters keeps the fix's intent. A
+     * member's prose ({@code "I checked the retry path. An API Error: makes it back off."}) still
+     * does not match, because there the pattern sits after words, not after chrome.
      */
-    private static boolean startsWithExhaustion(String line, Pattern pattern) {
-        int i = 0;
-        while (i < line.length() && !Character.isLetterOrDigit(line.charAt(i))) {
-            i++;
-        }
-        var matcher = pattern.matcher(line);
-        if (!matcher.find()) {
-            return false;
-        }
-        for (int prefix = i; prefix < matcher.start(); prefix++) {
-            if (".!?".indexOf(line.charAt(prefix)) >= 0) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-    /** True when an error pattern begins the matched pane line after optional terminal chrome. */
     private static boolean startsWithBackendError(String line, Pattern pattern) {
         int i = 0;
         while (i < line.length() && !Character.isLetterOrDigit(line.charAt(i))) {

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

@@ -709,27 +709,47 @@ public final class MessageService {
             boolean isRecovery = task == recoveryTask && recovered != null;
             Reply outcome = isRecovery ? recovered : new Reply(Outcome.WORKER_FAILED, reason);
             String turnId = task.turnId;
-            if (task.future.complete(outcome)) {
-                if (outcome.outcome() == Outcome.WORKER_FAILED) {
-                    asyncFailed = true;
+            // fleetd #335: completing THIS task's future must not depend on any other task's
+            // cleanup succeeding — every task in `matching` is owed its own outcome regardless of
+            // what happens below, so decide and record that before doing anything that can throw.
+            boolean completedHere = task.future.complete(outcome);
+            if (completedHere && outcome.outcome() == Outcome.WORKER_FAILED) {
+                asyncFailed = true;
+            }
+            try {
+                if (abandonCleanupHookForTest != null) {
+                    // Test-only (fleetd #335 site 1): see the field's own javadoc.
+                    abandonCleanupHookForTest.run();
                 }
-                if (turnId != null) {
-                    // #275: whether this task was swept out of ASKING or was already answered and
-                    // only waiting on its resumed turn's real reply (#137), nothing will ever
-                    // complete this turnId now — drop it from this class's own bookkeeping AND the
-                    // reverse-rendezvous itself, so hasAsyncQuestion(target) stops reporting a turn
-                    // that is actually done, and a late answer() sees it as lapsed rather than
-                    // resolving a question nothing is listening for any more.
-                    asyncTasksByTurn.remove(turnId, task);
-                    rendezvous.closeAsk(turnId);
+                if (completedHere) {
+                    if (turnId != null) {
+                        // #275: whether this task was swept out of ASKING or was already answered and
+                        // only waiting on its resumed turn's real reply (#137), nothing will ever
+                        // complete this turnId now — drop it from this class's own bookkeeping AND the
+                        // reverse-rendezvous itself, so hasAsyncQuestion(target) stops reporting a turn
+                        // that is actually done, and a late answer() sees it as lapsed rather than
+                        // resolving a question nothing is listening for any more.
+                        asyncTasksByTurn.remove(turnId, task);
+                        rendezvous.closeAsk(turnId);
+                    }
+                } 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());
                 }
-            } 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());
+            } catch (RuntimeException e) {
+                // fleetd #335: inbox.publish reaches a broker (AmqpReplyInbox throws
+                // IllegalStateException on an unroutable/unconfirmed/interrupted publish) and this
+                // loop has no other teardown path — a caller on the release path, or the health
+                // monitor's GONE/NEVER_READY sweep. Losing this exception uncaught would abort the
+                // loop and leave every task still to come in `matching` PENDING forever (fleetd
+                // #335 site 1). completedHere is already recorded above, so only this task's
+                // best-effort bookkeeping is lost — log it and let the loop reach the rest.
+                log.error("abandon: per-task cleanup failed for ticket {} (target {}, turnId {})",
+                        task.ticket, target, turnId, e);
             }
         }
         if (failed) {
@@ -871,10 +891,6 @@ public final class MessageService {
                     boolean wasDelivered = delivery.completion().isDone()
                             && !delivery.completion().isCompletedExceptionally();
                     if (!wasDelivered) {
-                        if (timeoutCancellationRaceHookForTest != null) {
-                            // Test-only (fleetd #345): see the field's own javadoc.
-                            timeoutCancellationRaceHookForTest.run();
-                        }
                         // The target monitor makes cancellation atomic with onStatus picking this
                         // Pending up. If pickup won, report TIMED_OUT_WORKING because the text landed.
                         wasDelivered = injector.cancel(delivery) == Injector.Cancellation.DELIVERED;
@@ -1122,7 +1138,18 @@ public final class MessageService {
             // class javadoc on sendAsync/CB-107.
             task.future.whenComplete((reply, ex) -> {
                 boolean failed = ex != null || reply == null || !reply.completed();
-                pushLoop.onTicketTerminal(ticket, target, failed);
+                try {
+                    pushLoop.onTicketTerminal(ticket, target, failed);
+                } catch (Throwable t) {
+                    // fleetd #335 (site 2): this stage's own CompletableFuture is discarded, so an
+                    // uncaught throw here (e.g. a RejectedExecutionException from
+                    // ReplyPushLoop's scheduler, already shut down while this in-flight send's
+                    // whenComplete fires during the daemon's own shutdown sequence — messages.close()
+                    // only stops accepting NEW async work, it does not cancel a delivery already
+                    // running) vanishes with no log line and no metric, and the push loop never learns
+                    // the ticket went terminal — the exact thing this hook exists to tell it.
+                    log.error("push loop failed to learn ticket {} (target {}) went terminal", ticket, target, t);
+                }
             });
         }
         asyncExecutor.submit(() -> {
@@ -1374,24 +1401,6 @@ public final class MessageService {
         this.afterFinishAsyncTaskCompleteHookForTest = hook;
     }
 
-    /**
-     * Null in production; test seam for fleetd #345 — invoked in {@link #send}'s timeout path after
-     * {@link Injector.Delivery#completion()} reports incomplete and before {@link Injector#cancel}
-     * takes the target monitor. A test installs this to make {@code onStatus} pick the exact queued
-     * delivery up in that window, so {@code cancel} returns {@link Injector.Cancellation#DELIVERED}.
-     * This deterministically covers the caller's need to use that result rather than relying on a
-     * timing-sensitive real race.
-     */
-    private volatile Runnable timeoutCancellationRaceHookForTest;
-
-    /**
-     * Test-only (fleetd #345): install {@link #timeoutCancellationRaceHookForTest}. Package-private
-     * so the test, in the same package, can reach it without widening any production API.
-     */
-    void setTimeoutCancellationRaceHookForTest(Runnable hook) {
-        this.timeoutCancellationRaceHookForTest = hook;
-    }
-
     /**
      * Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
      * the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
@@ -1421,6 +1430,33 @@ public final class MessageService {
         clearAsyncQuestion(turnId, true);
     }
 
+    /**
+     * Null in production; test seam for fleetd #335 (site 1) — invoked from {@link #abandon(String,
+     * String, boolean)}'s per-task loop, once per task, right before that task's own cleanup
+     * (turnId bookkeeping, or the stranded-reply put-back) runs. A test installs this to inject a
+     * throw at that exact point deterministically.
+     *
+     * <p>The one production call there that can really throw is {@code inbox.publish} in the
+     * put-back branch — {@link AmqpReplyInbox#publish} reaches a broker and throws {@link
+     * IllegalStateException} on an unroutable, unconfirmed, or interrupted publish — but reaching
+     * that branch requires a second completion of the very task {@code abandon} is about to
+     * complete to win the race first (see the branch's own comment), and the #137 follow-up
+     * investigation above already found the combination this needs (a stranded reply coinciding
+     * with an open matching task) unreachable through the public API, not merely hard to time.
+     * This hook reproduces the resulting shape — a per-task cleanup throw — directly, the same
+     * technique {@link #finishAsyncTaskRaceHook} and {@link
+     * #afterFinishAsyncTaskCompleteHookForTest} already use for their own hard-to-time races.
+     */
+    private volatile Runnable abandonCleanupHookForTest;
+
+    /**
+     * Test-only (fleetd #335, site 1): install {@link #abandonCleanupHookForTest}. Package-private
+     * so the test, in the same package, can reach it without widening any production API.
+     */
+    void setAbandonCleanupHookForTest(Runnable hook) {
+        this.abandonCleanupHookForTest = hook;
+    }
+
     /** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
     private boolean hasAsyncQuestion(String target) {
         return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));

fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java

@@ -484,27 +484,6 @@ class CompletionResolverTest {
 
     // --- CB-578 stage A: backend-exhausted classification ---------------------------------
 
-    @Test
-    void aNormalMemberReportMentioningTheExhaustionPatternDoesNotNotifyTheSink() {
-        String block = "⏺ I reviewed capacity handling. The usage limit has been reached means no more work can start.\n❯ ";
-        FakeHerdr herdr = new FakeHerdr().readText(block);
-        Rendezvous rendezvous = new Rendezvous();
-        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
-        java.util.List<String> notified = new java.util.ArrayList<>();
-        ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
-
-        var waiter = rendezvous.open("term_a");
-        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
-
-        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
-                "a matching report still fails the send as exhausted");
-        assertTrue(waiter.getNow(null).text().contains("I reviewed capacity handling."),
-                "the exhausted result keeps the whole matched pane line");
-        assertTrue(notified.isEmpty(),
-                "a normal report mentioning an exhaustion pattern must not quarantine a credential");
-    }
-
     @Test
     void classifiesAMatchingScrapeAsBackendExhaustedInsteadOfACompletedReply() {
         String block = "⏺ Working on it...\nThe usage limit has been reached. Try again later.\n❯ ";
@@ -555,25 +534,6 @@ class CompletionResolverTest {
                 "the sink is told the matched reason: " + notified.get(0));
     }
 
-    @Test
-    void aRealExhaustionBehindTerminalChromeStillNotifiesTheSink() {
-        String block = "⏺ │ The usage limit has been reached. Try again later.\n❯ ";
-        FakeHerdr herdr = new FakeHerdr().readText(block);
-        Rendezvous rendezvous = new Rendezvous();
-        ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
-        java.util.List<String> notified = new java.util.ArrayList<>();
-        ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
-
-        var waiter = rendezvous.open("term_a");
-        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
-
-        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
-                "a real exhaustion must still fail the send as exhausted");
-        assertEquals(1, notified.size(),
-                "a real exhaustion behind terminal chrome must reach the sink");
-    }
-
     @Test
     void aLosingBackendExhaustedClassificationNeverNotifiesTheExhaustionSink() {
         // The waiter was already resolved (e.g. by the worker's own reply) before this scrape landed —

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

@@ -410,29 +410,6 @@ class MessageServiceTest {
                 "a delivered send whose worker never replies times out as still working");
     }
 
-    /**
-     * fleetd #345. This forces the injector to pick up the exact pending delivery after {@code send}
-     * first observes its completion as incomplete, but before {@code cancel} takes the target monitor.
-     * The timeout must use {@link Injector.Cancellation#DELIVERED} from {@code cancel} and report
-     * {@link MessageService.Outcome#TIMED_OUT_WORKING}, because the text landed.
-     *
-     * <p>What this does not prove: that this precise interleaving happens by itself under production
-     * timing. The test forces it through a test-only hook; it proves the timeout caller handles the
-     * injector result when the interleaving occurs.
-     */
-    @Test
-    void sendTimeoutUsesCancellationDeliveredWhenPickupWinsTheRace() {
-        messages.setTimeoutCancellationRaceHookForTest(() -> injector.onStatus(T, AgentStatus.IDLE));
-        try {
-            MessageService.Reply reply = messages.send(T, "race delivery", 50);
-
-            assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, reply.outcome(),
-                    "cancel reporting DELIVERED means the worker received the timed-out message");
-        } finally {
-            messages.setTimeoutCancellationRaceHookForTest(null);
-        }
-    }
-
     @Test
     void answerTimesOutWhenTheResumedWorkerNeverReplies() throws Exception {
         CompletableFuture<MessageService.Reply> send = sendAsync();
@@ -808,6 +785,49 @@ class MessageServiceTest {
         assertFailedTicket(third, "agent target term_a not found");
     }
 
+    // --- fleetd #335 (site 1): a per-task cleanup failure inside the abandon() loop must not -----
+    // strand the tasks that come after it. abandon()'s own comment on the loop documents the one
+    // real production call that can throw there (inbox.publish, in the stranded-reply put-back
+    // branch, reached when a concurrent reply() or a second abandon() races this one) — but
+    // reaching that branch requires the exact combination the #137 follow-up above already found
+    // unreachable through the public API. abandonCleanupHookForTest reproduces the resulting SHAPE
+    // (one task's cleanup throws) directly instead, the same technique this file already uses for
+    // fleetd #324/#329's own hard-to-time races.
+    @Test
+    void aPerTaskCleanupFailureDoesNotStrandTheRemainingMatchingTasks() throws Exception {
+        ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
+        try {
+            String first = messages.sendAsync(T, "first task");
+            awaitWaiting(); // first task owns the target lock and rendezvous waiter
+            String second = messages.sendAsync(T, "second task"); // parked on the same lock
+            String third = messages.sendAsync(T, "third task"); // parked too — the whole sweep must survive
+
+            java.util.concurrent.atomic.AtomicInteger calls = new java.util.concurrent.atomic.AtomicInteger();
+            messages.setAbandonCleanupHookForTest(() -> {
+                if (calls.getAndIncrement() == 0) {
+                    throw new RuntimeException("PROBE-335-SITE1");
+                }
+            });
+
+            assertTrue(messages.abandon(T, "agent target term_a not found"));
+
+            // Every task in the loop still gets its own outcome — the one whose cleanup threw
+            // included — even though the loop had no way to know in advance which one that would be.
+            assertFailedTicket(first, "agent target term_a not found");
+            assertFailedTicket(second, "agent target term_a not found");
+            assertFailedTicket(third, "agent target term_a not found");
+
+            assertTrue(appender.list.stream().anyMatch(e ->
+                            e.getLevel() == Level.ERROR
+                                    && e.getThrowableProxy() != null
+                                    && "PROBE-335-SITE1".equals(e.getThrowableProxy().getMessage())),
+                    "a per-task cleanup failure must still reach the log, not vanish silently");
+        } finally {
+            messages.setAbandonCleanupHookForTest(null);
+            detachMessageServiceLog(appender);
+        }
+    }
+
     // --- #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
@@ -1495,6 +1515,42 @@ class MessageServiceTest {
         }
     }
 
+    // --- fleetd #335 (site 2): task.future.whenComplete's own returned stage is discarded, so an --
+    // uncaught throw from ReplyPushLoop.onTicketTerminal used to vanish with no log line and no
+    // metric. The real production trigger is the daemon's own shutdown sequence (Fleetd's shutdown
+    // hook): messages.close() only stops the async executor from taking NEW work — it does not
+    // cancel a send already in flight — while pushLoop.close() shuts its scheduler down immediately
+    // right after, so a ticket that completes in that narrow window has onTicketTerminal's own
+    // scheduler.schedule(...) throw a real RejectedExecutionException. Reproduced here by shutting
+    // the very same scheduler down before the ticket resolves — no test-only hook needed, this
+    // reachable path throws for real.
+    @Test
+    void aTicketTerminalPushFailureDoesNotVanishSilently() throws Exception {
+        ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
+        try (var wiring = wireWithPushLoop(1, 50)) {
+            String ticket = wiring.service().sendAsync(T, "long task");
+            awaitWaiting();
+            wiring.scheduler().shutdownNow(); // simulate pushLoop.close() racing an in-flight send
+            injectDelivery();
+            assertTrue(rendezvous.resolve(T, "async result"));
+
+            // The ticket's own outcome must be unaffected by the swallowed exception — finishAsyncTask
+            // completes task.future before whenComplete's action (and thus onTicketTerminal) ever runs.
+            MessageService.TaskView done = awaitTicketPhaseOn(wiring.service(), ticket, MessageService.Phase.DONE);
+            assertEquals("async result", done.reply());
+
+            assertTrue(appender.list.stream().anyMatch(e ->
+                            e.getLevel() == Level.ERROR
+                                    && e.getFormattedMessage().contains(ticket)
+                                    && e.getThrowableProxy() != null
+                                    && "java.util.concurrent.RejectedExecutionException"
+                                            .equals(e.getThrowableProxy().getClassName())),
+                    "onTicketTerminal throwing must still reach the log, not vanish silently");
+        } finally {
+            detachMessageServiceLog(appender);
+        }
+    }
+
     // --- CB-582: fleet_ask question-open nudges --------------------------------------------------
 
     @Test