fleetd#211: classify BACKEND_EXHAUSTED/BACKEND_ERROR from the raw scrape as a fallback

CompletionResolver.resolve() returned an empty-scrape failure before the
BACKEND_EXHAUSTED / BACKEND_ERROR classification ever ran, whenever
lastAssistantBlock() found no usable text — most commonly a pane with no ⏺
marker at all, whose boundary scan starts at the top of the raw screen and
breaks immediately on the first line of TUI chrome. Since BACKEND_EXHAUSTED
is the only caller of exhaustionSink, this meant an exhausted backend was
recorded as "produced nothing" instead of being quarantined.

Fix: run the same two classifications against the raw (untrimmed) scrape as
a fallback, only inside the empty-scrape failure branch. A pane that already
yields a usable assistant block never reaches this branch, so the existing
narrow match is unchanged. lastAssistantBlock stays the sole source of the
reply text; only classification ever consults the raw scrape.

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

@@ -237,11 +237,13 @@ public final class CompletionResolver implements TurnListener {
         }
         String tail;
         String assistantBlock = null;
+        String rawScrape = null;
         int originalLength = 0;
         boolean clipped = false;
         boolean scrapeFailed = false;
         try {
-            assistantBlock = lastAssistantBlock(agents.read(target, SCRAPE_SOURCE));
+            rawScrape = agents.read(target, SCRAPE_SOURCE);
+            assistantBlock = lastAssistantBlock(rawScrape);
             originalLength = assistantBlock.strip().length();
             clipped = originalLength > MAX_SCRAPE_CHARS;
             tail = clip(assistantBlock);
@@ -256,6 +258,20 @@ public final class CompletionResolver implements TurnListener {
         // member, so a caller (including a lead deciding whether to delegate again) can tell a lost
         // turn from a real empty answer.
         if (scrapeFailed || tail.isEmpty()) {
+            // fleetd#211: lastAssistantBlock() found nothing usable — most often a pane with no ⏺
+            // marker at all, whose boundary scan then starts at the top of the raw screen and breaks
+            // immediately on the first line of TUI chrome (╭, │, ❯, …). Before giving up as a lost
+            // turn, run the same exhaustion/backend-error classification against the RAW scrape as a
+            // fallback, ONLY here. A pane that already yielded a usable block never reaches this
+            // branch, so the narrow (trimmed) match on the normal path below is completely unchanged
+            // — zero new false positives there. Every pane this fallback examines was already headed
+            // for the empty-scrape failure, so a wrong label here is strictly less bad than silently
+            // losing an exhaustion signal: the alternative outcome is already a failure, just one that
+            // never quarantines the credential. lastAssistantBlock stays the source of the reply
+            // TEXT everywhere else; only classification ever consults the raw scrape, and only here.
+            if (rawScrape != null && classifyRawScrapeFallback(target, turn, waiter, rawScrape)) {
+                return;
+            }
             fail(target, turn, emptyScrapeReason(target, scrapeFailed));
             return;
         }
@@ -313,6 +329,38 @@ public final class CompletionResolver implements TurnListener {
         }
     }
 
+    /**
+     * fleetd#211: the raw-scrape fallback classification, run only when {@link #lastAssistantBlock}
+     * found nothing usable (see the call site in {@link #resolve}). Mirrors the two classifications
+     * the normal path already applies to the trimmed assistant block — exhaustion first, then the
+     * narrow {@link #BACKEND_ERROR} pattern — against {@code raw} instead, and reports whether one of
+     * them handled the turn (resolved the waiter or failed it) so the caller skips the empty-scrape
+     * failure. Never runs on the normal (non-empty-block) path, and never touches the reply text.
+     */
+    private boolean classifyRawScrapeFallback(String target, InFlight turn,
+                                               CompletableFuture<Rendezvous.Resolution> waiter, String raw) {
+        Pattern exhausted = exhaustedPatterns.patternFor(target);
+        String matchedLine = exhausted == null ? null : firstMatchingLine(raw, exhausted);
+        if (matchedLine != null) {
+            String reason = "backend exhausted (usage limit): " + matchedLine;
+            if (rendezvous.resolveExhausted(waiter, reason)) {
+                inFlight.remove(target, turn);
+                log.warn("completion for {} classified BACKEND_EXHAUSTED from the raw scrape (no "
+                        + "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.
+                exhaustionSink.onExhausted(target, reason);
+            }
+            return true;
+        }
+        String backendError = firstMatchingLine(raw, BACKEND_ERROR);
+        if (backendError != null) {
+            fail(target, turn, "member " + target + " ended on a backend error: " + backendError);
+            return true;
+        }
+        return false;
+    }
+
     /** Synchronous fail (the unit-testable core of {@link #onTurnFailed}). */
     void fail(String target, InFlight turn) {
         fail(target, turn, null);

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

@@ -633,6 +633,109 @@ class CompletionResolverTest {
                 "the failure carries the rest of the pane, not only the matched line: " + reason);
     }
 
+    // --- fleetd#211: raw-scrape fallback classification when there is no usable assistant block ---
+
+    @Test
+    void anExhaustionLineWithNoMarkerAndLeadingChromeIsClassifiedFromTheRawScrapeAndNotifiesTheSink() {
+        // No ⏺ anywhere, and the first visible line is TUI chrome (╭). lastAssistantBlock's boundary
+        // scan starts at the top of the raw screen and breaks immediately, so the trimmed block is "".
+        // The fix: fall back to matching the RAW scrape so this doesn't get lost as an empty scrape.
+        String block = """
+                ╭──────────────────────────────────────╮
+                The usage limit has been reached. Try again later.
+                """;
+        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) -> 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));
+
+        assertTrue(waiter.isDone(), "a raw-scrape match still resolves the blocked send");
+        assertEquals(Rendezvous.Kind.BACKEND_EXHAUSTED, waiter.getNow(null).kind(),
+                "classified from the raw scrape even though the trimmed block was empty");
+        assertEquals(1, notified.size(),
+                "the sink is the whole point of this ticket — it must be notified: " + notified);
+        assertTrue(notified.get(0).startsWith("term_a: "), "the sink is told which target exhausted");
+        assertTrue(notified.get(0).contains("The usage limit has been reached"),
+                "the sink is told the matched reason: " + notified.get(0));
+    }
+
+    @Test
+    void aBackendErrorLineWithNoMarkerAndLeadingChromeIsClassifiedFromTheRawScrape() {
+        String block = """
+                ╭──────────────────────────────────────╮
+                API Error: 400 invalid request body
+                """;
+        FakeHerdr herdr = new FakeHerdr().readText(block);
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
+
+        var waiter = rendezvous.open("term_a");
+        resolver.resolve("term_a", new CompletionResolver.InFlight(waiter, null));
+
+        assertTrue(waiter.isDone(), "a raw-scrape backend-error match still resolves the blocked send");
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "classified BACKEND_ERROR from the raw scrape even though the trimmed block was empty");
+        assertTrue(waiter.getNow(null).text().contains("API Error: 400 invalid request body"),
+                "the failure carries the matched line: " + waiter.getNow(null).text());
+    }
+
+    @Test
+    void anOrdinaryPaneWithANormalAssistantBlockIsUnaffectedByTheRawScrapeFallback() {
+        // Pin: on a pane that already yields a usable block, the fallback branch is never reached —
+        // same outcome, same text, sink not called — even though the raw screen around the marker
+        // would itself match the configured exhausted pattern.
+        String block = "The usage limit has been reached, but this is a leading TUI line above the "
+                + "marker.\n⏺ complete report\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) -> 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));
+
+        assertTrue(waiter.isDone());
+        assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
+                "unchanged: a usable assistant block never reaches the raw-scrape fallback");
+        assertEquals("complete report", waiter.getNow(null).text(), "the reply text is unaffected");
+        assertTrue(notified.isEmpty(), "the fallback never runs, so the sink is never called");
+    }
+
+    @Test
+    void aGenuinelyEmptyScrapeStillFailsAsEmptyAndNeverNotifiesTheSink() {
+        // The false-positive pin: no exhaustion or backend-error text anywhere on the pane (just
+        // chrome, no marker) — the raw-scrape fallback must not manufacture a classification, and
+        // the sink must stay untouched.
+        String block = """
+                ╭──────────────────────────────────────╮
+                │ >                                    │
+                ╰──────────────────────────────────────╯
+                """;
+        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) -> 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));
+
+        assertTrue(waiter.isDone(), "an empty scrape must still resolve the send, not hang");
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "no exhaustion or backend-error text anywhere ⇒ this stays the ordinary empty-scrape failure");
+        assertTrue(waiter.getNow(null).text().toLowerCase().contains("empty"),
+                "the failure still says the scrape was empty: " + waiter.getNow(null).text());
+        assertTrue(notified.isEmpty(), "a genuinely empty pane must never quarantine a credential");
+    }
+
     @Test
     void coverageIsOffWhenNoProfileHasAPatternConfigured() {
         assertEquals("off (no profile has an exhaustedPattern configured; profiles: [terra])",