diff --git a/bridged/src/main/java/dev/ltms/bridged/health/FleetHealthMonitor.java b/bridged/src/main/java/dev/ltms/bridged/health/FleetHealthMonitor.java index c0d808c..79e4f84 100644 --- a/bridged/src/main/java/dev/ltms/bridged/health/FleetHealthMonitor.java +++ b/bridged/src/main/java/dev/ltms/bridged/health/FleetHealthMonitor.java @@ -9,6 +9,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.concurrent.ScheduledExecutorService; @@ -26,6 +27,10 @@ public final class FleetHealthMonitor { private final LongSupplier clock; private final long intervalSeconds; private final Map priors = new HashMap<>(); + private final Map states = new HashMap<>(); + + // These facts need the evidence publishers introduced by later M4 units. They are not negatives. + private static final boolean NOT_YET_OBSERVED = false; public FleetHealthMonitor(AgentControl agents, Supplier> roster, MessageService messages, ScheduledExecutorService scheduler, LongSupplier clock, long intervalSeconds) { @@ -47,21 +52,53 @@ public final class FleetHealthMonitor { // Package-private so tests can run one tick without waiting. void tick() { - List agentsNow = agents.list(); // Exactly one list call for this complete observation. - Map live = new HashMap<>(); - for (Agent agent : agentsNow) live.put(agent.terminalId(), agent); - for (MemberSession session : roster.get()) { // One in-memory roster snapshot for the same tick. - Agent agent = live.get(session.terminalId()); - AgentStatus status = agent == null ? AgentStatus.UNKNOWN : agent.status(); - boolean accepted = messages.hasAcceptedDelivery(session.terminalId()); - HealthSnapshot snapshot = new HealthSnapshot(session.state(), status, accepted, false, - messages.hasInboxMessage(session.terminalId()), agent != null, false, false, - false, false, false, false); - HealthDecision decision = decide(snapshot, priors.getOrDefault(session.terminalId(), HealthPrior.NONE), - clock.getAsLong()); - priors.put(session.terminalId(), decision.prior()); + try { + List agentsNow = agents.list(); // Exactly one list call for this complete observation. + List rosterNow = roster.get(); // One in-memory roster snapshot for this tick. + Map live = new HashMap<>(); + for (Agent agent : agentsNow) live.put(agent.terminalId(), agent); + HashSet current = new HashSet<>(); + for (MemberSession session : rosterNow) { + current.add(session.terminalId()); + Agent agent = live.get(session.terminalId()); + AgentStatus status = agent == null ? AgentStatus.UNKNOWN : agent.status(); + boolean accepted = messages.hasAcceptedDelivery(session.terminalId()); + HealthSnapshot snapshot = new HealthSnapshot(session.state(), status, accepted, NOT_YET_OBSERVED, + messages.hasInboxMessage(session.terminalId()), agent != null, NOT_YET_OBSERVED, + NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED, NOT_YET_OBSERVED); + HealthDecision decision = decide(snapshot, priors.getOrDefault(session.terminalId(), HealthPrior.NONE), + clock.getAsLong()); + priors.put(session.terminalId(), decision.prior()); + reportTransition(session.terminalId(), decision.state()); + } + priors.keySet().retainAll(current); + states.keySet().retainAll(current); + } catch (Throwable error) { + // A list failure is health evidence, and must never kill the monitor's only scheduler task. + log.warn("fleet health collection failed; will retry next tick", error); + } finally { + if (!scheduler.isShutdown()) { + scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS); + } } - scheduler.schedule(this::tick, intervalSeconds, TimeUnit.SECONDS); + } + + void reportTransition(String target, HealthState next) { + HealthState previous = states.put(target, next); + if (previous == next) return; + if (fault(next)) { + log.warn("fleet health member={} state={} previous={}", target, next, previous); + } else if (previous != null && fault(previous)) { + log.info("fleet health member={} recovered state={} previous={}", target, next, previous); + } + } + + private static boolean fault(HealthState state) { + return switch (state) { + case NEVER_READY, GONE, TURN_BOUNDARY_LOST, ERROR_ON_SCREEN, STALL_SUSPECTED, + MUTE, REPLY_STRANDED, DELEGATION_ORPHANED, CONTROL_LINK_DOWN -> true; + default -> false; + }; } public static String coverage(boolean enabled, boolean notificationConfigured) { diff --git a/bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java b/bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java index e133ad5..4d90b07 100644 --- a/bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java +++ b/bridged/src/test/java/dev/ltms/bridged/health/FleetHealthMonitorTest.java @@ -1,5 +1,8 @@ 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; @@ -12,6 +15,7 @@ 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; @@ -38,4 +42,40 @@ class FleetHealthMonitorTest { 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 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); + } + } }