Merge #473: fleet_profiles and fleet_list report which attempt a quarantine is on (fleetd #466 item 2)

The escalating cooldown landed in 789b6a8, but the only thing an operator
could see was quarantinedForSeconds. A long number does not say whether
this is the first exhaustion or the fifth, and after escalation those look
the same from outside: 3600 seconds could be a big base cooldown or a
credential on its fourth strike.

Both reports now carry quarantineAttempt beside quarantinedForSeconds.

The part that matters is HOW, not that the field exists.
BackendQuarantine#status(credentialId) does ONE quarantines.get() and
returns both values from the same QuarantineState. It is not two accessors
that a caller pairs up. That is the pattern
CompositePeerLauncher.modelGateState() already documents: a report that
reads a different source from the behaviour it describes will eventually
disagree with it, and the disagreement is invisible because both halves
look right on their own.

This is reporting only. Escalation, the ceiling and the quiet-gap reset are
untouched. The flat two-argument constructor still reports a real growing
attempt count even though its cooldown stays flat - which is the honest
answer, since the streak is real whether or not the cooldown uses it.

The worker's report was lost: its fleet_reply never arrived and its inbox
was empty. Nothing was lost with it, because the brief required pushing
first and putting the report in the PR body. That is the second time that
practice has saved a turn.

Numbers and my own mutation battery are on the ticket, measured on this
merge commit.

fleetd/src/main/java/dev/ltms/fleet/mcp/FleetMcp.java

@@ -1298,6 +1298,13 @@ public final class FleetMcp {
      * (the backend text that triggered the most recent quarantine of that credential, omitted when
      * none is known) — so a lead can see WHICH model to turn off and WHY, without reading the
      * daemon log.
+     *
+     * <p>fleetd #466 scope item 2: each {@code quarantined} row also names {@code
+     * quarantineAttempt} — 1 for a first-time exhaustion, 2 for the second in a row, and so on — so
+     * an operator sees "this is the 5th time" instead of inferring it from {@code
+     * quarantinedForSeconds} alone. Read off {@link BackendQuarantine#status(String)}, the same
+     * one-call accessor {@code quarantinedForSeconds} itself comes from here (see its doc) — never a
+     * separately derived count.
      */
     public static Map<String, Object> profilesView(PeerLauncher workers, QuarantineSource quarantine, OutageSource outage) {
         Map<String, Object> result = new LinkedHashMap<>();
@@ -1310,10 +1317,14 @@ public final class FleetMcp {
             exhaustionDetectionArmed.put(profile, quarantine.exhaustedPatternArmed().apply(profile));
             String credentialId = quarantine.credentialIdFor().apply(profile);
             if (credentialId != null) {
-                quarantine.quarantine().remainingSeconds(credentialId).ifPresent(remaining -> {
+                // fleetd #466 scope item 2: quarantinedForSeconds and quarantineAttempt come off the
+                // ONE BackendQuarantine#status(credentialId) call — never a second, independent read
+                // for the attempt count — so the two can never disagree about which streak this is.
+                quarantine.quarantine().status(credentialId).ifPresent(status -> {
                     Map<String, Object> row = new LinkedHashMap<>();
                     row.put("credentialId", credentialId);
-                    row.put("quarantinedForSeconds", remaining);
+                    row.put("quarantinedForSeconds", status.remainingSeconds());
+                    row.put("quarantineAttempt", status.repeatCount());
                     String model = quarantine.modelFor().apply(profile);
                     if (model != null && !model.isBlank()) {
                         row.put("model", model);
@@ -1753,10 +1764,13 @@ public final class FleetMcp {
         }
         String credentialId = quarantine.credentialIdFor().apply(profile);
         if (credentialId != null) {
-            quarantine.quarantine().remainingSeconds(credentialId).ifPresent(remaining -> {
+            // fleetd #466 scope item 2: same one-call status() read as profilesView above — see its
+            // comment for why this must not become two separate lookups.
+            quarantine.quarantine().status(credentialId).ifPresent(status -> {
                 row.put("free", 0);
                 row.put("credentialId", credentialId);
-                row.put("quarantinedForSeconds", remaining);
+                row.put("quarantinedForSeconds", status.remainingSeconds());
+                row.put("quarantineAttempt", status.repeatCount());
             });
         }
         String outageCredentialId = outage.credentialIdFor().apply(profile);

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

@@ -3,6 +3,7 @@ package dev.ltms.fleet.placement;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.LongSupplier;
@@ -205,6 +206,49 @@ public final class BackendQuarantine {
         return remaining > 0 ? OptionalLong.of(toSecondsRoundedUp(remaining)) : OptionalLong.empty();
     }
 
+    /**
+     * Remaining seconds together with which consecutive exhaustion this is (fleetd #466 scope item
+     * 2) — {@code repeatCount} 1 for a first occurrence, 2 for the second in a row, and so on; see
+     * {@link #quarantine}'s class-doc Mechanism section for exactly when a call continues a streak
+     * versus starts a fresh one.
+     *
+     * <p><strong>Read together, off the one {@link QuarantineState} entry {@link #quarantine} itself
+     * wrote</strong> — a single {@code quarantines.get(credentialId)}, never a separate lookup or a
+     * value re-derived from {@code remainingSeconds} (e.g. inverting {@link
+     * #escalatedCooldownNanos}). That inversion is not just extra work to avoid: once a streak has
+     * hit {@code maxCooldownNanos}, every further consecutive exhaustion reports the identical
+     * cooldown, so a derivation that starts from the cooldown value cannot tell the 4th repeat from
+     * the 9th — only the stored {@code repeatCount} can. This is the same rule {@code
+     * CompositePeerLauncher.modelGateState()} documents for its own gate/report pair: the report
+     * reads the exact accessor the behaviour reads, so it can never disagree with what actually
+     * happened (the fleetd #404/#422 lesson). {@code fleet_profiles}/{@code fleet_list}/{@code GET
+     * /profiles} all call this — never {@link #remainingSeconds} plus a second, independent count —
+     * for exactly that reason.
+     *
+     * @return empty when {@code credentialId} is not currently quarantined (including on {@link
+     *         #none()}, which quarantines nothing)
+     */
+    public Optional<Status> status(String credentialId) {
+        QuarantineState state = quarantines.get(credentialId);
+        if (state == null) {
+            return Optional.empty();
+        }
+        long remaining = state.deadlineNanos() - nowNanos.getAsLong();
+        return remaining > 0
+                ? Optional.of(new Status(toSecondsRoundedUp(remaining), state.repeatCount()))
+                : Optional.empty();
+    }
+
+    /**
+     * @param remainingSeconds seconds left on the quarantine, identical to {@link
+     *                          #remainingSeconds(String)}'s answer for the same credential at the
+     *                          same instant
+     * @param repeatCount       1 for a first occurrence, 2 for the second consecutive one, etc. —
+     *                          see {@link #status(String)}
+     */
+    public record Status(long remainingSeconds, int repeatCount) {
+    }
+
     /**
      * Every currently-quarantined credential id and its remaining seconds (CB-578 stage B fleet
      * reporting) — expired entries are never included. Not pruned from the backing map here: it stays

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetMcpTest.java

@@ -574,6 +574,38 @@ class FleetMcpTest {
         assertTrue(out.contains("\"quarantinedForSeconds\":1800"), out);
     }
 
+    /**
+     * fleetd #466 scope item 2: {@code fleet_profiles} must carry the repeat count beside the
+     * remaining seconds, and the two must come off the one {@link BackendQuarantine#status} call so
+     * they can never disagree about which streak this is (see {@code profilesView}'s javadoc).
+     */
+    @Test
+    void profilesReportsQuarantineAttemptBesideRemainingSeconds() {
+        FakeHerdr h = new FakeHerdr();
+        java.util.concurrent.atomic.AtomicLong now = new java.util.concurrent.atomic.AtomicLong(0L);
+        BackendQuarantine quarantine = BackendQuarantine.withEscalation(now::get, TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai"); // attempt 1: 1800s
+        now.set(TimeUnit.MINUTES.toNanos(30));
+        quarantine.quarantine("shared-openai"); // attempt 2: 3600s
+        FleetMcp.QuarantineSource source = new FleetMcp.QuarantineSource(
+                profile -> "ltms-local".equals(profile) ? "shared-openai" : null, quarantine);
+        McpSchema.CallToolResult res = FleetMcp.profiles(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), source);
+        String out = textOf(res);
+        assertTrue(out.contains("\"quarantinedForSeconds\":3600"), out);
+        assertTrue(out.contains("\"quarantineAttempt\":2"), out);
+    }
+
+    /** A never-quarantined profile must not carry {@code quarantineAttempt} either. */
+    @Test
+    void profilesOmitsQuarantineAttemptWhenNotQuarantined() {
+        FakeHerdr h = new FakeHerdr();
+        McpSchema.CallToolResult res = FleetMcp.profiles(
+                workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")), FleetMcp.QuarantineSource.none());
+        String out = textOf(res);
+        assertFalse(out.contains("quarantineAttempt"), out);
+    }
+
     /** fleetd #201 Unit 5: {@code coolingOff} is a SEPARATE map from {@code quarantined}. */
     @Test
     void profilesReportsACoolingOffCredentialInASeparateMap() {
@@ -1121,6 +1153,35 @@ class FleetMcpTest {
         assertTrue(out.contains("\"quarantinedForSeconds\":1200"), out);
     }
 
+    /**
+     * fleetd #466 scope item 2: {@code fleet_list}'s capacity rows (CB-583: they reuse quarantine)
+     * must carry {@code quarantineAttempt} beside {@code quarantinedForSeconds}, off the same
+     * {@link BackendQuarantine#status} call as {@code fleet_profiles} -- see {@code capacityView}'s
+     * comment pointing back to {@code profilesView}.
+     */
+    @Test
+    void capacityRowReportsQuarantineAttemptBesideRemainingSeconds() {
+        FakeHerdr h = new FakeHerdr();
+        SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
+        java.util.concurrent.atomic.AtomicLong now = new java.util.concurrent.atomic.AtomicLong(0L);
+        BackendQuarantine quarantine = BackendQuarantine.withEscalation(now::get, TimeUnit.MINUTES.toNanos(20));
+        quarantine.quarantine("shared-openai"); // attempt 1
+        now.set(TimeUnit.MINUTES.toNanos(20));
+        quarantine.quarantine("shared-openai"); // attempt 2
+        now.set(TimeUnit.MINUTES.toNanos(60));
+        quarantine.quarantine("shared-openai"); // attempt 3
+        FleetMcp.QuarantineSource source = new FleetMcp.QuarantineSource(
+                profile -> "terra".equals(profile) ? "shared-openai" : null, quarantine);
+
+        String out = textOf(FleetMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
+                sessions, null, new FleetMcp.CapacitySource(profile -> 0, profile -> 2,
+                () -> Set.of("terra"), () -> 0), new FleetMcp.HealthCoverageSource(() -> "off"),
+                source, Map.of(), ""));
+
+        assertTrue(out.contains("\"credentialId\":\"shared-openai\""), out);
+        assertTrue(out.contains("\"quarantineAttempt\":3"), out);
+    }
+
     @Test
     void everyProfileSharingTheQuarantinedCredentialReportsZeroFree() {
         FakeHerdr h = new FakeHerdr();

fleetd/src/test/java/dev/ltms/fleet/placement/BackendQuarantineTest.java

@@ -229,4 +229,149 @@ class BackendQuarantineTest {
         assertEquals(OptionalLong.of(3600L), q.remainingSeconds("shared-openai"),
                 "the default multiplier must be 2.0");
     }
+
+    // --- fleetd #466 scope item 2: status() reports repeatCount beside remainingSeconds ------------
+
+    @Test
+    void statusReportsAttemptOneForAFirstOccurrenceNeverAbsentOrZero() {
+        BackendQuarantine q = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30), 2.0,
+                TimeUnit.HOURS.toNanos(6));
+        q.quarantine("shared-openai");
+
+        BackendQuarantine.Status status = q.status("shared-openai").orElseThrow();
+        assertEquals(1800L, status.remainingSeconds());
+        assertEquals(1, status.repeatCount(),
+                "a first-ever occurrence must report attempt 1, not 0 or absent -- 1 means unambiguously "
+                        + "'the first time', where 0 would be indistinguishable from a bug that forgot to count");
+    }
+
+    @Test
+    void statusIsAbsentWhenTheCredentialIsNotQuarantined() {
+        BackendQuarantine q = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30), 2.0,
+                TimeUnit.HOURS.toNanos(6));
+        assertTrue(q.status("shared-openai").isEmpty());
+    }
+
+    /**
+     * The acceptance criterion's strong form: the reported {@code repeatCount} must match the exact
+     * step the cooldown's own growth implies, read off {@link BackendQuarantine#status}'s single
+     * call -- not two independent reads that happen to agree in this easy case.
+     */
+    @Test
+    void statusReportsTheGrowingAttemptCountAlongsideTheEscalatingCooldown() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.SECONDS.toNanos(600), 2.0,
+                TimeUnit.SECONDS.toNanos(2400));
+
+        q.quarantine("shared-openai"); // 1st: 600s, attempt 1
+        BackendQuarantine.Status first = q.status("shared-openai").orElseThrow();
+        assertEquals(600L, first.remainingSeconds());
+        assertEquals(1, first.repeatCount());
+
+        now.set(TimeUnit.SECONDS.toNanos(600));
+        q.quarantine("shared-openai"); // 2nd: 1200s, attempt 2
+        BackendQuarantine.Status second = q.status("shared-openai").orElseThrow();
+        assertEquals(1200L, second.remainingSeconds());
+        assertEquals(2, second.repeatCount());
+
+        now.set(TimeUnit.SECONDS.toNanos(1800));
+        q.quarantine("shared-openai"); // 3rd: 2400s (at the ceiling), attempt 3
+        BackendQuarantine.Status third = q.status("shared-openai").orElseThrow();
+        assertEquals(2400L, third.remainingSeconds());
+        assertEquals(3, third.repeatCount());
+    }
+
+    /**
+     * Once the cooldown hits its ceiling, every further consecutive exhaustion reports the SAME
+     * {@code remainingSeconds} -- so a {@code repeatCount} re-derived from the cooldown value (e.g.
+     * inverting {@code cooldownNanos * multiplier^(n-1)}) could not tell attempt 4 from attempt 9;
+     * only the stored counter can. This is the scenario that makes "read the count off a second,
+     * independent computation" provably wrong rather than just risky.
+     */
+    @Test
+    void repeatCountKeepsGrowingPastTheCeilingEvenThoughTheCooldownStaysFlat() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.SECONDS.toNanos(600), 2.0,
+                TimeUnit.SECONDS.toNanos(2400));
+
+        long t = 0L;
+        for (int attempt = 1; attempt <= 5; attempt++) {
+            now.set(t);
+            q.quarantine("shared-openai");
+            BackendQuarantine.Status status = q.status("shared-openai").orElseThrow();
+            assertEquals(attempt, status.repeatCount(),
+                    "attempt " + attempt + " must be reported as exactly " + attempt
+                            + ", not collapsed to whatever attempt first reached the ceiling");
+            if (attempt >= 3) {
+                assertEquals(2400L, status.remainingSeconds(), "attempt " + attempt + " must be capped");
+            }
+            t += status.remainingSeconds(); // land exactly on the next deadline: still the same streak
+        }
+    }
+
+    @Test
+    void aQuietGapResetsTheReportedAttemptCountToOneToo() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.SECONDS.toNanos(600), 2.0,
+                TimeUnit.SECONDS.toNanos(2400));
+
+        q.quarantine("shared-openai");
+        now.set(TimeUnit.SECONDS.toNanos(600));
+        q.quarantine("shared-openai");
+        now.set(TimeUnit.SECONDS.toNanos(1800));
+        q.quarantine("shared-openai"); // attempt 3
+        assertEquals(3, q.status("shared-openai").orElseThrow().repeatCount());
+
+        now.set(TimeUnit.SECONDS.toNanos(20_000)); // long quiet gap
+        q.quarantine("shared-openai");
+        assertEquals(1, q.status("shared-openai").orElseThrow().repeatCount(),
+                "a reset streak must report attempt 1 again, matching the reset base cooldown");
+    }
+
+    @Test
+    void escalatingOneCredentialsAttemptCountDoesNotAffectAnother() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.SECONDS.toNanos(600), 2.0,
+                TimeUnit.SECONDS.toNanos(2400));
+
+        q.quarantine("shared-openai");
+        now.set(TimeUnit.SECONDS.toNanos(600));
+        q.quarantine("shared-openai");
+        now.set(TimeUnit.SECONDS.toNanos(1800));
+        q.quarantine("shared-openai"); // 3-in-a-row streak on this credential only
+
+        q.quarantine("another-credential");
+        assertEquals(1, q.status("another-credential").orElseThrow().repeatCount(),
+                "an unrelated credential's attempt count must stay at 1, unaffected by a sibling's streak");
+    }
+
+    @Test
+    void noneReportsNoStatusForAnything() {
+        BackendQuarantine q = BackendQuarantine.none();
+        q.quarantine("shared-openai"); // no-op on none(), same as every other mutator
+
+        assertTrue(q.status("shared-openai").isEmpty(),
+                "none() quarantines nothing, so it must report no status at all -- never a fabricated "
+                        + "attempt count for a credential that was never actually quarantined");
+    }
+
+    /** The flat (non-escalating) two-argument constructor must still report a real, growing count. */
+    @Test
+    void aFlatTwoArgumentInstanceStillReportsAGrowingAttemptCountEvenThoughTheCooldownStaysFlat() {
+        AtomicLong now = new AtomicLong(0L);
+        BackendQuarantine q = new BackendQuarantine(now::get, TimeUnit.SECONDS.toNanos(600));
+
+        q.quarantine("shared-openai");
+        BackendQuarantine.Status first = q.status("shared-openai").orElseThrow();
+        assertEquals(600L, first.remainingSeconds());
+        assertEquals(1, first.repeatCount());
+
+        now.set(TimeUnit.SECONDS.toNanos(100)); // still inside the 1st quarantine
+        q.quarantine("shared-openai");
+        BackendQuarantine.Status second = q.status("shared-openai").orElseThrow();
+        assertEquals(600L, second.remainingSeconds(),
+                "the flat constructor's cooldown must stay exactly the base length regardless of the streak");
+        assertEquals(2, second.repeatCount(),
+                "the flat constructor still counts the real streak -- it just does not scale the cooldown by it");
+    }
 }