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

@@ -6,12 +6,14 @@ import dev.ltms.fleet.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongSupplier;
 import java.util.regex.Pattern;
 
 /**
@@ -61,6 +63,17 @@ public final class CompletionResolver implements TurnListener {
     /** Cap the scraped tail so a long transcript can't return an unbounded blob. */
     static final int MAX_SCRAPE_CHARS = 4000;
 
+    /**
+     * fleetd#164: the floor below which a {@code BUSY -> DONE} transition cannot be real work. A
+     * backend that rejects a turn outright (e.g. an HTTP 400 from the model, before the worker read
+     * a single file or produced a token) drives the exact same confirmed {@code working -> idle}
+     * transition a genuine completion does — just in about a second instead of the many seconds a
+     * real turn costs. {@link #onTurnComplete} cannot tell those two cases apart from the transition
+     * alone, so a turn that settles inside this floor is treated as a crash signature and resolved
+     * as a failure, never as a (possibly empty) success.
+     */
+    public static final long MIN_TURN_NANOS = Duration.ofSeconds(2).toNanos();
+
     private static final String CLIPPED_PANE_TAIL_MARKER =
             "[Pane tail clipped: member did not call fleet_reply.]";
 
@@ -68,6 +81,7 @@ public final class CompletionResolver implements TurnListener {
     private final Rendezvous rendezvous;
     private final ExhaustedPatternLookup exhaustedPatterns;
     private final ExhaustionSink exhaustionSink;
+    private final LongSupplier nowNanos;
 
     /**
      * Per-target record of the turn currently in flight: the exact {@link Rendezvous} waiter its
@@ -79,8 +93,21 @@ public final class CompletionResolver implements TurnListener {
      * reference: a completion scrape equal to it means the worker produced no new output (the previous
      * turn's wind-down sampled as this boundary), so it is suppressed. Overwritten on each delivery;
      * cleared when the turn resolves. Package-private so tests can capture and replay a specific turn.
+     *
+     * <p>{@code deliveredAtNanos} (fleetd#164) is the {@link #nowNanos} reading taken at delivery —
+     * the other half of the {@link #MIN_TURN_NANOS} floor check, compared against a fresh reading at
+     * resolution time.
      */
-    record InFlight(CompletableFuture<Rendezvous.Resolution> waiter, String baseline) {
+    record InFlight(CompletableFuture<Rendezvous.Resolution> waiter, String baseline, long deliveredAtNanos) {
+
+        /**
+         * Convenience for tests exercising scrape/suppression logic that don't care about turn
+         * timing: back-dates the delivery far enough that {@link #MIN_TURN_NANOS} can never fire.
+         * Not used by production code — {@link #captureBaseline} always records a real reading.
+         */
+        InFlight(CompletableFuture<Rendezvous.Resolution> waiter, String baseline) {
+            this(waiter, baseline, Long.MIN_VALUE / 2);
+        }
     }
 
     private final ConcurrentHashMap<String, InFlight> inFlight = new ConcurrentHashMap<>();
@@ -97,10 +124,25 @@ public final class CompletionResolver implements TurnListener {
      */
     public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns,
                               ExhaustionSink exhaustionSink) {
+        this(agents, rendezvous, exhaustedPatterns, exhaustionSink, System::nanoTime);
+    }
+
+    /**
+     * Test constructor with an injectable clock (fleetd#164), matching the {@code LongSupplier}
+     * pattern {@link dev.ltms.fleet.session.SessionManager} and {@link dev.ltms.fleet.msg.MessageService}
+     * already use: lets a test place a turn's delivery and its resolution at an exact, controllable
+     * distance apart around the {@link #MIN_TURN_NANOS} floor, without a real sleep. Public (rather
+     * than package-private like those two) because callers that wire a full {@code MessageService}
+     * fixture — e.g. {@code MessageServiceTest} — construct this resolver directly from another
+     * package.
+     */
+    public CompletionResolver(AgentControl agents, Rendezvous rendezvous, ExhaustedPatternLookup exhaustedPatterns,
+                       ExhaustionSink exhaustionSink, LongSupplier nowNanos) {
         this.agents = agents;
         this.rendezvous = rendezvous;
         this.exhaustedPatterns = Objects.requireNonNull(exhaustedPatterns, "exhaustedPatterns");
         this.exhaustionSink = Objects.requireNonNull(exhaustionSink, "exhaustionSink");
+        this.nowNanos = Objects.requireNonNull(nowNanos, "nowNanos");
     }
 
     @Override
@@ -130,7 +172,7 @@ public final class CompletionResolver implements TurnListener {
             baseline = null; // fail open: no baseline ⇒ no suppression
             log.debug("delivery baseline for {} failed: {}", target, e.getMessage());
         }
-        inFlight.put(target, new InFlight(waiter, baseline));
+        inFlight.put(target, new InFlight(waiter, baseline, nowNanos.getAsLong()));
     }
 
     /** The turn currently baselined for {@code target}, or {@code null} — a test hook for the captureBaseline path. */
@@ -176,6 +218,15 @@ public final class CompletionResolver implements TurnListener {
             inFlight.remove(target, turn);
             return;
         }
+        // fleetd#164: a BUSY -> DONE transition inside the floor cannot be real work — it's a crash
+        // signature (e.g. a backend HTTP 400 before the worker did anything), not a fast answer. Fail
+        // it before spending a scrape on the ordinary path; the reason still carries whatever is on
+        // screen, since that is usually the backend's own error.
+        long elapsedNanos = nowNanos.getAsLong() - turn.deliveredAtNanos();
+        if (elapsedNanos < MIN_TURN_NANOS) {
+            fail(target, turn, tooFastReason(target, elapsedNanos));
+            return;
+        }
         String tail;
         String assistantBlock = null;
         int originalLength = 0;
@@ -187,20 +238,25 @@ public final class CompletionResolver implements TurnListener {
             clipped = originalLength > MAX_SCRAPE_CHARS;
             tail = clip(assistantBlock);
         } catch (RuntimeException e) {
-            // The worker finished but we couldn't read its screen — still resolve the send so the
-            // caller unblocks; an empty tail beats hanging until the caller's timeout.
-            log.warn("completion scrape for {} failed; resolving with an empty tail: {}",
-                    target, e.getMessage());
+            log.warn("completion scrape for {} failed: {}", target, e.getMessage());
             tail = "";
             scrapeFailed = true;
         }
+        // fleetd#164: a scrape nobody could read, and a scrape that read cleanly but produced nothing,
+        // both used to resolve the send as a SUCCESS carrying "" — indistinguishable from a worker that
+        // genuinely finished with nothing to say. That is the defect: fail loudly instead, naming the
+        // 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()) {
+            fail(target, turn, emptyScrapeReason(target, scrapeFailed));
+            return;
+        }
         // Misattribution guard (CB-115): if the scrape is byte-identical to the pane content at
         // delivery, this turn produced no new output — the boundary belongs to the previous turn's
         // wind-down (common on rapid back-to-back sends). Suppress rather than resolve the send with
         // a stale answer; the real fleet_reply (or a later genuine completion) resolves it instead.
-        // A scrape that failed to read is exempt — an empty tail there is "couldn't see", not "no change".
         String baseline = turn.baseline();
-        if (!scrapeFailed && baseline != null && baseline.equals(tail)) {
+        if (baseline != null && baseline.equals(tail)) {
             log.debug("suppressing misattributed completion for {} (no output change since delivery)",
                     target);
             return; // keep the in-flight record: a later genuine completion still needs it
@@ -208,21 +264,19 @@ public final class CompletionResolver implements TurnListener {
         // CB-578 stage A: a turn that ended with no fleet_reply AND whose scrape matches the
         // backend's configured usage-limit pattern is a refusal, not an answer. Classify it as
         // BACKEND_EXHAUSTED rather than handing the caller a scrape that reads like a real reply.
-        if (!scrapeFailed) {
-            Pattern exhausted = exhaustedPatterns.patternFor(target);
-            String matchedLine = exhausted == null ? null : firstMatchingLine(assistantBlock, 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 (no fleet_reply; scrape "
-                            + "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.
-                    exhaustionSink.onExhausted(target, reason);
-                }
-                return;
+        Pattern exhausted = exhaustedPatterns.patternFor(target);
+        String matchedLine = exhausted == null ? null : firstMatchingLine(assistantBlock, 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 (no fleet_reply; scrape "
+                        + "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.
+                exhaustionSink.onExhausted(target, reason);
             }
+            return;
         }
         String completion = clipped ? tail + "\n" + CLIPPED_PANE_TAIL_MARKER : tail;
         if (rendezvous.resolveCompletion(waiter, completion)) {
@@ -272,6 +326,36 @@ public final class CompletionResolver implements TurnListener {
         }
     }
 
+    /**
+     * fleetd#164: the failure reason for a turn that settled inside {@link #MIN_TURN_NANOS} — names
+     * the member and both timings, and appends whatever the pane shows (usually the backend's own
+     * error) so the caller sees the cause, not just "it failed".
+     */
+    private String tooFastReason(String target, long elapsedNanos) {
+        String scrape;
+        try {
+            scrape = clip(agents.read(target, SCRAPE_SOURCE));
+        } catch (RuntimeException e) {
+            scrape = "";
+        }
+        String reason = String.format(
+                "member %s went BUSY -> DONE in %dms (floor %dms) — too fast to be real work, most "
+                        + "likely a backend error before any work started",
+                target, elapsedNanos / 1_000_000, MIN_TURN_NANOS / 1_000_000);
+        return scrape.isBlank() ? reason : reason + ": " + scrape;
+    }
+
+    /**
+     * fleetd#164: the failure reason for a scrape that produced zero characters — names the member
+     * and says plainly that the turn produced nothing, so a caller (a lead deciding whether to
+     * delegate again included) never mistakes a lost turn for a genuinely empty reply.
+     */
+    private static String emptyScrapeReason(String target, boolean scrapeFailed) {
+        return "member " + target + " turn completed with an empty scrape (0 chars) — "
+                + (scrapeFailed ? "its pane could not be read; " : "")
+                + "treating as a lost turn, not a real answer";
+    }
+
     /**
      * The first line of {@code text} matching {@code pattern}, stripped — the CB-578 stage A
      * evidence carried in a {@code BACKEND_EXHAUSTED} reason so the operator sees the real refusal

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

@@ -197,10 +197,15 @@ class CompletionResolverTest {
     void resolvesSynchronouslyBeforePostTurnContextClearing() {
         FakeHerdr herdr = new FakeHerdr().readText("⏺ previous answer\n❯ ");
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
+        // fleetd#164: an injectable clock, so this real (non-crash) turn lands outside MIN_TURN_NANOS
+        // — captureBaseline and resolveBeforePostAction below run back-to-back with no real delay.
+        long[] clock = {1_000_000_000L};
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
         var waiter = rendezvous.open("term_a");
         resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
         herdr.readText("⏺ answer that /clear would erase\n❯ ");
+        clock[0] += CompletionResolver.MIN_TURN_NANOS + 1; // this turn took longer than the floor
 
         resolver.resolveBeforePostAction("term_a");
 
@@ -219,7 +224,11 @@ class CompletionResolverTest {
         String longBlock = "⏺ " + "x".repeat(CompletionResolver.MAX_SCRAPE_CHARS + 500) + "\n❯ ";
         FakeHerdr herdr = new FakeHerdr().readText(longBlock);
         Rendezvous rendezvous = new Rendezvous();
-        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
+        // fleetd#164: an injectable clock so captureBaseline and resolve (back-to-back, no real
+        // delay) don't trip the too-fast-turn floor — this test is about the suppression guard, not timing.
+        long[] clock = {1_000_000_000L};
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
 
         var waiter = rendezvous.open("term_a");        // a send is blocked on this turn
         resolver.captureBaseline("term_a", new TurnToken("term_a", waiter)); // baseline is the clipped >cap block
@@ -227,6 +236,7 @@ class CompletionResolverTest {
         assertEquals(CompletionResolver.MAX_SCRAPE_CHARS, turn.baseline().length(),
                 "the delivery baseline is clipped to the same cap resolve() applies to the tail");
 
+        clock[0] += CompletionResolver.MIN_TURN_NANOS + 1; // outside the floor
         resolver.resolve("term_a", turn);              // scrape unchanged → clipped tail == baseline → suppress
 
         assertFalse(waiter.isDone(),
@@ -249,12 +259,13 @@ class CompletionResolverTest {
     }
 
     @Test
-    void resolvesWhenTheScrapeItselfFailsEvenWithABaselinePresent() {
-        // The most important branch of the CB-115 guard: a failed read means the resolver could not
-        // SEE the screen — "couldn't see", not "no change". It must still resolve the send (an empty
-        // tail beats hanging until the caller's timeout), even though a baseline was captured. The
-        // baseline here is "" (an empty pane at delivery), so without the !scrapeFailed clause the
-        // byte-identical guard would wrongly match the empty tail and suppress.
+    void aFailedScrapeResolvesAsAFailureEvenWithABaselinePresent() {
+        // fleetd#164: this test used to assert that a failed read resolved the send as a SUCCESS
+        // carrying an empty string ("an empty tail beats hanging until the caller's timeout") — that
+        // was the bug this ticket fixes: a lost turn and a genuine empty answer looked identical to
+        // every caller. This test encoded the bug and is changed here: a failed read must fail the
+        // send instead, naming the member, whether or not a baseline was captured (the baseline here
+        // is "", an empty pane at delivery — proof this isn't the CB-115 misattribution path either).
         FakeHerdr herdr = new FakeHerdr().healthy(false); // agent.read throws HerdrException
         Rendezvous rendezvous = new Rendezvous();
         CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
@@ -265,8 +276,82 @@ class CompletionResolverTest {
 
         assertTrue(waiter.isDone(),
                 "a failed scrape must still resolve the send, not hang until the caller's timeout");
-        assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind());
-        assertEquals("", waiter.getNow(null).text(), "the tail is empty because the screen was unreadable");
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "a failed read is a lost turn, not a successful empty reply");
+        assertTrue(waiter.getNow(null).text().contains("term_a"),
+                "the failure names the member: " + waiter.getNow(null).text());
+        assertTrue(waiter.getNow(null).text().contains("could not be read"),
+                "the failure explains the scrape could not be read: " + waiter.getNow(null).text());
+    }
+
+    // --- fleetd#164: an empty (but readable) scrape must never resolve as a success --------------
+
+    @Test
+    void anEmptyScrapeResolvesAsAFailureNamingTheMember() {
+        // The core defect: a scrape that read CLEANLY but produced zero characters used to resolve
+        // the send as a SUCCESS carrying "" — indistinguishable, to every caller, from a worker that
+        // genuinely finished with nothing to say. A lost turn must never look like a real empty reply.
+        FakeHerdr herdr = new FakeHerdr().readText("");
+        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(), "an empty scrape must still resolve the send, not hang");
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind(),
+                "an empty scrape is a lost turn, not a successful empty reply");
+        assertTrue(waiter.getNow(null).text().contains("term_a"),
+                "the failure names the member: " + waiter.getNow(null).text());
+        assertTrue(waiter.getNow(null).text().toLowerCase().contains("empty"),
+                "the failure says the scrape was empty: " + waiter.getNow(null).text());
+    }
+
+    // --- fleetd#164: a BUSY -> DONE transition inside the floor is a crash, not a fast answer ------
+
+    @Test
+    void aBusyToDoneTransitionInsideTheFloorResolvesAsAFailure() {
+        // The exact fleetd#164 scenario: the backend returned an HTTP 400 before the worker did
+        // anything, and the member went BUSY -> DONE in ~1s. That transition alone is indistinguishable
+        // from a genuine (if unusually fast) completion, so the resolver leans on the floor to catch it.
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ HTTP 400: invalid request\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        long[] clock = {10_000_000_000L};
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
+
+        var waiter = rendezvous.open("term_a");
+        var turn = new CompletionResolver.InFlight(waiter, null, clock[0]); // delivered "now"
+        clock[0] += CompletionResolver.MIN_TURN_NANOS - 1; // 1ns inside the floor
+
+        resolver.resolve("term_a", turn);
+
+        assertTrue(waiter.isDone(), "a suspiciously fast turn must still resolve (as a failure)");
+        assertEquals(Rendezvous.Kind.FAILED, waiter.getNow(null).kind());
+        assertTrue(waiter.getNow(null).text().contains("term_a"),
+                "the failure names the member: " + waiter.getNow(null).text());
+        assertTrue(waiter.getNow(null).text().contains("HTTP 400"),
+                "the failure carries whatever was on screen: " + waiter.getNow(null).text());
+    }
+
+    @Test
+    void aBusyToDoneTransitionJustOutsideTheFloorResolvesNormally() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ a real, if quick, answer\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        long[] clock = {10_000_000_000L};
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
+
+        var waiter = rendezvous.open("term_a");
+        var turn = new CompletionResolver.InFlight(waiter, null, clock[0]); // delivered "now"
+        clock[0] += CompletionResolver.MIN_TURN_NANOS + 1; // 1ns outside the floor
+
+        resolver.resolve("term_a", turn);
+
+        assertTrue(waiter.isDone());
+        assertEquals(Rendezvous.Kind.COMPLETION, waiter.getNow(null).kind(),
+                "a turn that took longer than the floor resolves normally");
+        assertEquals("a real, if quick, answer", waiter.getNow(null).text());
     }
 
     // --- CB-115/CB-116 fail guard: an already-done or absent waiter is left alone ---------

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

@@ -33,8 +33,17 @@ class MessageServiceTest {
     private final FakeHerdr herdr = new FakeHerdr().readText("BUILD GREEN: 391 files");
     private final AgentControl agents = new AgentControl(herdr);
     private final Rendezvous rendezvous = new Rendezvous();
+    /**
+     * fleetd#164: this fixture drives a delivery and its completion back-to-back with no real time
+     * between them, so the real clock would trip {@link CompletionResolver#MIN_TURN_NANOS} on every
+     * completion-fallback test here. An ever-advancing fake clock stands in for the model-latency and
+     * herdr round-trips a real turn would spend, so each delivery-then-resolve pair still lands
+     * outside the floor.
+     */
+    private final java.util.concurrent.atomic.AtomicLong resolverClock = new java.util.concurrent.atomic.AtomicLong();
     private final CompletionResolver completion =
-            new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none());
+            new CompletionResolver(agents, rendezvous, ExhaustedPatternLookup.none(), ExhaustionSink.none(),
+                    () -> resolverClock.addAndGet(CompletionResolver.MIN_TURN_NANOS + 1));
     private final Injector injector = new Injector(agents, completion);
     private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
     private final MessageService messages = new MessageService(agents, injector, rendezvous, inbox);