fleetd #273: validate exhaustedPattern regex at load, like errorPattern

FleetConfig.rejectMalformedErrorPattern only compiled errorPattern eagerly
at config load. exhaustedPattern was compiled unguarded in Fleetd.main,
so profiles.<name>.exhaustedPattern: "[" passed load() and then crashed
the whole daemon at boot with a raw PatternSyntaxException naming neither
the profile nor the key.

Rename the validator to rejectMalformedProfilePatterns and extend it to
also compile every non-blank exhaustedPattern, reporting
profiles.<name>.exhaustedPattern ("<value>"): <message> in the same style
as errorPattern. Both keys are collected and reported together from a
single load. Fleetd.java's compile site is left as-is per scope — it is
now safe because load already rejects a bad value.

Added tests covering: a bad exhaustedPattern is refused; a bad pattern in
each key is reported together in one message; valid patterns still load;
a blank/absent exhaustedPattern is ignored.

fleetd/src/main/java/dev/ltms/fleet/config/FleetConfig.java

@@ -1506,7 +1506,7 @@ public record FleetConfig(
             rejectDuplicateMemberSlots(yaml);
             rejectNegativeMaxLoad(yaml);
             rejectAutoCompactWindowOutOfRange(yaml);
-            rejectMalformedErrorPattern(yaml);
+            rejectMalformedProfilePatterns(yaml);
             rejectUnknownKind(yaml);
             rejectUnknownAuthMode(yaml);
             rejectUnknownPlacement(yaml);
@@ -1856,20 +1856,26 @@ public record FleetConfig(
     }
 
     /**
-     * Reject a profile whose {@code errorPattern} (fleetd #201 Unit 5) is not a valid Java regex,
-     * naming the profile, the key, and the parser's own message.
+     * Reject a profile whose {@code errorPattern} (fleetd #201 Unit 5) or {@code exhaustedPattern}
+     * (CB-578 stage A) is not a valid Java regex, naming the profile, the key, and the parser's own
+     * message.
      *
-     * <p>Unset/{@code null} means "use {@code CompletionResolver}'s built-in {@code (?i)\bAPI
-     * Error\s*:} compatibility pattern" and passes silently. A profile that DOES set the key gets it
-     * compiled once at daemon startup ({@code Fleetd.main}, mirroring {@code exhaustedPattern}) — an
-     * uncaught {@link java.util.regex.PatternSyntaxException} there crashes startup without naming
-     * which profile or key is at fault. Validate eagerly here instead, at config load, the same
-     * "fail loud at load, not lazily later" reasoning as {@link #rejectAutoCompactWindowOutOfRange}.
+     * <p>Unset/{@code null} means, for {@code errorPattern}, "use {@code CompletionResolver}'s
+     * built-in {@code (?i)\bAPI Error\s*:} compatibility pattern", and for {@code exhaustedPattern},
+     * "opt out of that classification" — either way it passes silently. A profile that DOES set
+     * either key gets it compiled once at daemon startup ({@code Fleetd.main}) — an uncaught
+     * {@link java.util.regex.PatternSyntaxException} there crashes startup without naming which
+     * profile or key is at fault (fleetd #273: this happened for {@code exhaustedPattern}, which had
+     * no validator here even though its sibling {@code errorPattern} did). Validate eagerly here
+     * instead, at config load, the same "fail loud at load, not lazily later" reasoning as
+     * {@link #rejectAutoCompactWindowOutOfRange}. Both keys are checked from a single load, and any
+     * failures from either are collected together into one message.
      *
      * @param yaml the raw config text
-     * @throws IllegalStateException when any profile's {@code errorPattern} fails to compile
+     * @throws IllegalStateException when any profile's {@code errorPattern} or
+     *                                {@code exhaustedPattern} fails to compile
      */
-    static void rejectMalformedErrorPattern(String yaml) {
+    static void rejectMalformedProfilePatterns(String yaml) {
         Map<?, ?> raw;
         try {
             raw = YAML.readValue(yaml, Map.class);
@@ -1884,18 +1890,20 @@ public record FleetConfig(
             if (!(e.getValue() instanceof Map<?, ?> p)) {
                 continue;
             }
-            if (!(p.get("errorPattern") instanceof String pattern) || pattern.isBlank()) {
-                continue;
-            }
-            try {
-                Pattern.compile(pattern);
-            } catch (PatternSyntaxException ex) {
-                bad.add("profiles." + e.getKey() + ".errorPattern (\"" + pattern + "\"): " + ex.getMessage());
+            for (String key : List.of("errorPattern", "exhaustedPattern")) {
+                if (!(p.get(key) instanceof String pattern) || pattern.isBlank()) {
+                    continue;
+                }
+                try {
+                    Pattern.compile(pattern);
+                } catch (PatternSyntaxException ex) {
+                    bad.add("profiles." + e.getKey() + "." + key + " (\"" + pattern + "\"): " + ex.getMessage());
+                }
             }
         }
         bad.sort(String::compareTo);
         if (!bad.isEmpty()) {
-            throw new IllegalStateException("refusing to start: malformed errorPattern — "
+            throw new IllegalStateException("refusing to start: malformed pattern — "
                     + String.join("; ", bad));
         }
     }

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

@@ -319,10 +319,12 @@ public final class FleetMcp {
                 };
         BiFunction<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> pollHandler =
                 (exchange, req) -> {
-                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
-                    if (denied != null) return denied;
                     Map<String, Object> a = req.arguments();
-                    return poll(messages, str(a, "ticket"), str(a, "target"));
+                    String target = str(a, "target");
+                    // The action depends on the ARGUMENTS, not on the tool name -- see pollAction.
+                    McpSchema.CallToolResult denied = deny(exchange, pollAction(target), target);
+                    if (denied != null) return denied;
+                    return poll(messages, str(a, "ticket"), target);
                 };
         // CB-307 Increment 3: per-msgId ack (not needed in v1 but supported by the inbox).
         // Acking removes a reply from the inbox, so it is a drain, not a read.
@@ -724,6 +726,34 @@ public final class FleetMcp {
         return text("delivered to peer lead " + coordId + " (msgId " + msg.msgId() + ")");
     }
 
+    /**
+     * Which authorization action a {@code fleet_poll} call needs, decided by its arguments
+     * (fleetd #272).
+     *
+     * <p>{@code fleet_poll} is <strong>two operations behind one tool name</strong>. With {@code
+     * ticket} it observes an async delegation and changes nothing, which is a {@link
+     * Authz.Action#READ}. With {@code target} it calls {@link MessageService#drainReplies} on that
+     * session -- the replies are removed from the inbox and a second call returns nothing -- so it
+     * is a {@link Authz.Action#DRAIN}, the same gate {@code fleet_ack} already uses for removing a
+     * single message, and the same one the REST path uses at {@code FleetApp.drainReplies}.
+     *
+     * <p>Until this method existed the handler passed a constant {@code READ} for both branches.
+     * {@code READ} is open to every authenticated role, so any worker could read a peer's id out of
+     * {@code fleet_list} and destroy the replies that peer had queued for the primary. The gate
+     * failed open, and it did so because the required action is a function of the arguments while
+     * the handler chose it before looking at them.
+     *
+     * <p>The choice lives in this method, and not inline in the handler, so that a test can assert
+     * the mapping the handler actually uses. {@code FleetMcpAuthzTest} already checked every
+     * {@link Authz.Action} against every {@link Role} and passed throughout -- it tested the policy
+     * table, which was correct, while the defect was in which action the caller handed it.
+     *
+     * @param target the {@code target} argument of the call, or {@code null}/blank when absent
+     */
+    static Authz.Action pollAction(String target) {
+        return isBlank(target) ? Authz.Action.READ : Authz.Action.DRAIN;
+    }
+
     /** {@code fleet_poll}: check an async delegation by ticket, or drain a worker's inbox by target. */
     static McpSchema.CallToolResult poll(MessageService messages, String ticket, String target) {
         if (!isBlank(target)) {

fleetd/src/test/java/dev/ltms/fleet/config/FleetConfigTest.java

@@ -155,6 +155,90 @@ class FleetConfigTest {
         assertTrue(e.getMessage().contains("errorPattern"), "the offending key is named: " + e.getMessage());
     }
 
+    // ── fleetd #273: exhaustedPattern gets the same load-time validation as its sibling errorPattern ──
+
+    @Test
+    void aProfileWithAMalformedExhaustedPatternIsRejectedAtLoadNamingTheProfileAndKey(@TempDir Path dir)
+            throws Exception {
+        Path f = dir.resolve("malformed-exhausted-pattern.yaml");
+        Files.writeString(f, """
+                profiles:
+                  ltms-local:
+                    baseUrl: http://gx00.gw:8000
+                    exhaustedPattern: "["
+                """);
+
+        IllegalStateException e = assertThrows(IllegalStateException.class, () -> FleetConfig.load(f));
+        assertTrue(e.getMessage().contains("ltms-local"), "the offending profile is named: " + e.getMessage());
+        assertTrue(e.getMessage().contains("exhaustedPattern"), "the offending key is named: " + e.getMessage());
+    }
+
+    @Test
+    void aMalformedErrorPatternAndAMalformedExhaustedPatternAreBothReportedFromOneLoad(@TempDir Path dir)
+            throws Exception {
+        Path f = dir.resolve("both-malformed.yaml");
+        Files.writeString(f, """
+                profiles:
+                  sonnet:
+                    baseUrl: http://gx00.gw:8000
+                    errorPattern: "(unterminated["
+                  terra:
+                    baseUrl: http://gx01.gw:8000
+                    exhaustedPattern: "["
+                """);
+
+        IllegalStateException e = assertThrows(IllegalStateException.class, () -> FleetConfig.load(f));
+        assertTrue(e.getMessage().contains("sonnet"), "the errorPattern profile is named: " + e.getMessage());
+        assertTrue(e.getMessage().contains("errorPattern"), e.getMessage());
+        assertTrue(e.getMessage().contains("terra"), "the exhaustedPattern profile is named: " + e.getMessage());
+        assertTrue(e.getMessage().contains("exhaustedPattern"), e.getMessage());
+    }
+
+    @Test
+    void validErrorPatternAndExhaustedPatternBothLoadFine(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("both-valid.yaml");
+        Files.writeString(f, """
+                profiles:
+                  ltms-local:
+                    baseUrl: http://gx00.gw:8000
+                    errorPattern: "credential outage"
+                    exhaustedPattern: "usage limit has been reached"
+                """);
+
+        FleetConfig.Profile w = FleetConfig.load(f).profiles().get("ltms-local");
+        assertEquals("credential outage", w.errorPattern());
+        assertEquals("usage limit has been reached", w.exhaustedPattern());
+    }
+
+    @Test
+    void aBlankExhaustedPatternNormalizesToNullJustLikeUnset(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("blank-exhausted-pattern.yaml");
+        Files.writeString(f, """
+                profiles:
+                  ltms-local:
+                    baseUrl: http://gx00.gw:8000
+                    exhaustedPattern: "   "
+                """);
+
+        FleetConfig.Profile w = FleetConfig.load(f).profiles().get("ltms-local");
+        assertNull(w.exhaustedPattern());
+        assertFalse(w.hasExhaustedPattern());
+    }
+
+    @Test
+    void aProfileWithNoExhaustedPatternLoadsFine(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-exhausted-pattern.yaml");
+        Files.writeString(f, """
+                profiles:
+                  ltms-local:
+                    baseUrl: http://gx00.gw:8000
+                """);
+
+        FleetConfig.Profile w = FleetConfig.load(f).profiles().get("ltms-local");
+        assertNull(w.exhaustedPattern());
+        assertFalse(w.hasExhaustedPattern());
+    }
+
     @Test
     void withProfileCarriesErrorPatternThrough(@TempDir Path dir) throws Exception {
         Path f = dir.resolve("with-profile-error-pattern.yaml");

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

@@ -179,6 +179,53 @@ class FleetMcpAuthzTest {
                 "no CallerResolver supplied ⇒ authorization not enforced (legacy behaviour)");
     }
 
+    // --- which action each tool hands the gate (fleetd #272) ------------------------------------
+
+    /**
+     * fleetd #272: {@code fleet_poll{target}} drains a session's reply inbox, so it needs
+     * {@link Authz.Action#DRAIN} -- not the {@link Authz.Action#READ} the handler passed for both
+     * of its branches until this ticket.
+     *
+     * <p>This asserts against {@link FleetMcp#pollAction}, the method the handler itself calls, so
+     * the handler holds no separate copy of the rule that this test could miss. Every other test in
+     * this class checks the policy table (is a worker allowed to DRAIN?) and all of them passed for
+     * the whole time the defect was live -- the table was right, the action fed to it was wrong.
+     */
+    @Test
+    void pollingByTargetIsADrainAndPollingByTicketIsARead() {
+        assertEquals(Authz.Action.DRAIN, FleetMcp.pollAction("term_b"),
+                "poll by target removes the replies — that is a drain, not an observation");
+        assertEquals(Authz.Action.READ, FleetMcp.pollAction(null),
+                "poll by ticket changes nothing");
+        assertEquals(Authz.Action.READ, FleetMcp.pollAction("  "),
+                "a blank target is an absent target");
+    }
+
+    @Test
+    void aWorkerMayNotDrainAnotherSessionsInboxByPolling() {
+        FleetMcp m = mcp(true);
+
+        assertNotNull(m.denyFor(WORKER_A, FleetMcp.pollAction("term_b"), "term_b"),
+                "a worker draining a peer's inbox would destroy replies queued for the primary");
+        assertNotNull(m.denyFor(ARCH_DESIGN, FleetMcp.pollAction("term_b"), "term_b"),
+                "an architect has no lifecycle rights either — same gate as fleet_ack");
+        assertNull(m.denyFor(PRIMARY, FleetMcp.pollAction("term_b"), "term_b"),
+                "collecting a held reply is the primary's job");
+    }
+
+    /**
+     * The tightening must not close the branch that legitimately serves non-primary callers: an
+     * architect may {@code fleet_send}, so it owns tickets and must be able to poll them.
+     */
+    @Test
+    void pollingAnOwnTicketStaysOpenToWorkersAndArchitects() {
+        FleetMcp m = mcp(true);
+
+        assertNull(m.denyFor(WORKER_A, FleetMcp.pollAction(null), null));
+        assertNull(m.denyFor(ARCH_DESIGN, FleetMcp.pollAction(null), null),
+                "an architect delegates with wait:false, so it must be able to poll its ticket");
+    }
+
     // --- identity reconstruction from the transport context ------------------------------------
 
     @Test