fleetd #466 scope item 2: report the quarantine repeat count, not only the seconds

Add BackendQuarantine#status(credentialId) -> Optional<Status>, a single
QuarantineState read that answers both remainingSeconds and repeatCount
together -- the same "one accessor" pattern CompositePeerLauncher.
modelGateState() already uses, so the two facts can never disagree.

fleet_profiles/REST GET /profiles and fleet_list's capacity rows
(FleetMcp.profilesView/capacityView) now call status() instead of
remainingSeconds() and add a "quarantineAttempt" field beside
"quarantinedForSeconds": 1 for a first occurrence, 2 for the second
in a row, and so on.

No change to the escalation, ceiling, or reset logic itself -- this
unit is reporting only.

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

@@ -1269,6 +1269,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<>();
@@ -1281,10 +1288,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);
@@ -1724,10 +1735,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");
+    }
 }