Merge CB-573: dormant fleet health monitor (M4 unit 1)

An opt-in whole-fleet observer, separate from the 250ms delivery poller.
One AgentControl.list and one roster snapshot per tick, joined and fed to
the FleetHealth classifier, because a fault is a disagreement between the
two views at the same instant. Absent a health: block nothing is built
and no herdr call is made.

Adds bridge_list healthCoverage: off, detection-only, or full. Detection
is deliberately separate from notification, so a single-lead setup with
no webhook still gets detection and is told its coverage is partial
rather than being refused.

Two review fixes worth naming. tick() rescheduled itself as its last
statement with no try/catch, and a ScheduledExecutorService does not
re-run a task that threw — so the first agents.list failure would have
stopped health permanently and silently, which is exactly when the
control link is down. It now catches Throwable and reschedules in a
finally. And the snapshot fields this unit cannot supply are the named
constant NOT_YET_OBSERVED rather than bare false literals, because false
means no fault to this classifier.

bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java

@@ -0,0 +1,81 @@
+package dev.ltms.bridged.health;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.ltms.bridged.herdr.AgentControl;
+import dev.ltms.bridged.herdr.FakeHerdr;
+import dev.ltms.bridged.inject.Injector;
+import dev.ltms.bridged.msg.InMemoryReplyInbox;
+import dev.ltms.bridged.msg.MessageService;
+import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.session.SessionManager;
+import dev.ltms.bridged.member.ClaudeCodeLauncher;
+import dev.ltms.bridged.herdr.WorkspaceControl;
+import dev.ltms.bridged.guard.SubscriptionGuard;
+import dev.ltms.bridged.config.BridgedConfig;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class FleetHealthMonitorTest {
+    @Test void oneTickUsesOneFleetListForAnyRosterSize() {
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        BridgedConfig.Profile profile = new BridgedConfig.Profile("test", "http://test:1", null,
+                null, null, null, null, null, null, null, null, null);
+        ClaudeCodeLauncher launcher = new ClaudeCodeLauncher(agents, new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("test")), Map.of("test", profile), "test", _ -> "token");
+        SessionManager sessions = new SessionManager(launcher);
+        sessions.acquire("test", null, null, null);
+        sessions.acquire("test", null, null, null);
+        herdr.calls.clear();
+        MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox());
+        var scheduler = Executors.newSingleThreadScheduledExecutor();
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, sessions::roster, messages, scheduler, () -> 1, 60);
+        monitor.tick();
+        monitor.stop();
+        assertEquals(1, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
+    }
+
+    @Test void failedTickDoesNotStopTheNextTick() {
+        FakeHerdr herdr = new FakeHerdr().healthy(false);
+        AgentControl agents = new AgentControl(herdr);
+        var scheduler = Executors.newSingleThreadScheduledExecutor();
+        FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
+                new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
+                scheduler, () -> 1, 60);
+        monitor.tick();
+        herdr.healthy(true);
+        monitor.tick();
+        monitor.stop();
+        assertEquals(2, herdr.calls.stream().filter(call -> call.method().equals("agent.list")).count());
+    }
+
+    @Test void faultTransitionLogsOnlyOnceUntilItChanges() {
+        Logger logger = (Logger) LoggerFactory.getLogger(FleetHealthMonitor.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+        logger.addAppender(appender);
+        try {
+            FakeHerdr herdr = new FakeHerdr();
+            AgentControl agents = new AgentControl(herdr);
+            var scheduler = Executors.newSingleThreadScheduledExecutor();
+            FleetHealthMonitor monitor = new FleetHealthMonitor(agents, java.util.List::of,
+                    new MessageService(agents, new Injector(agents), new Rendezvous(), new InMemoryReplyInbox()),
+                    scheduler, () -> 1, 60);
+            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
+            monitor.reportTransition("term_a", HealthState.TURN_BOUNDARY_LOST);
+            monitor.stop();
+            assertEquals(1, appender.list.stream().filter(event -> event.getFormattedMessage()
+                    .contains("member=term_a state=TURN_BOUNDARY_LOST")).count());
+        } finally {
+            logger.detachAppender(appender);
+        }
+    }
+}

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpAuthzTest.java

@@ -72,7 +72,7 @@ class BridgeMcpAuthzTest {
                 new PrimaryRegistry(null),
                 enforce ? CallerResolver.withLeadsAndMembers(identity, false, null,
                 Map::of, new MemberRegistry(null)) : null,
-                metrics, BridgeMcp.CapacitySource.none());
+                metrics, BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"));
         return mcp;
     }
 

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java

@@ -474,7 +474,7 @@ class BridgeMcpTest {
         sessions.acquire("ltms-local", null, null, null);
         McpSchema.CallToolResult res = BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 sessions, null, new BridgeMcp.CapacitySource(profile -> 2, profile -> 2,
-                () -> Set.of("ltms-local"), () -> 0), Map.of(), "");
+                () -> Set.of("ltms-local"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), "");
         String out = textOf(res);
         assertTrue(out.contains("\"maxLoad\":2"), out);
         assertTrue(out.contains("\"live\":2"), out);
@@ -487,7 +487,7 @@ class BridgeMcpTest {
         SessionManager sessions = new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")));
         String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 sessions, null, new BridgeMcp.CapacitySource(profile -> 0, profile -> 2,
-                () -> Set.of("terra"), () -> 0), Map.of(), ""));
+                () -> Set.of("terra"), () -> 0), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
         assertTrue(out.contains("\"profile\":\"terra\""), out);
         assertTrue(out.contains("\"live\":0"), out);
         assertTrue(out.contains("\"free\":2"), out);
@@ -499,7 +499,7 @@ class BridgeMcpTest {
         FakeHerdr h = new FakeHerdr();
         String out = textOf(BridgeMcp.listFleet(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw")),
                 new SessionManager(workerService(h, "http://gx00.gw:8000", Set.of("gx00.gw"))), null,
-                BridgeMcp.CapacitySource.none(), Map.of(), ""));
+                BridgeMcp.CapacitySource.none(), new BridgeMcp.HealthCoverageSource(() -> "off"), Map.of(), ""));
         assertFalse(out.contains("\"capacity\":"), out);
     }