Merge CB-551: idle-lead heartbeat

bridged/src/test/java/dev/ltms/bridged/config/BridgedConfigTest.java

@@ -132,6 +132,7 @@ class BridgedConfigTest {
                 leaders: {}
                 architects: {}
                 leadScan: {}
+                leadHeartbeat: {}
                 placement: fixed
                 auth: {}
                 """).isEmpty(), "the known-key set must not drift from the record components");
@@ -180,6 +181,51 @@ class BridgedConfigTest {
         assertEquals(30, scan.intervalSeconds());
     }
 
+    // ── CB-551: the idle-lead heartbeat ─────────────────────────────────────────────────────────
+
+    /**
+     * Acceptance (f): absent config ⇒ the heartbeat is OFF completely. It spends the operator's
+     * subscription on its own initiative, so an upgraded daemon must never switch it on unasked.
+     */
+    @Test
+    void leadHeartbeatIsOffUnlessTheBlockIsPresent(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-heartbeat.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\n");
+
+        assertNull(BridgedConfig.load(f).leadHeartbeat(),
+                "no leadHeartbeat: block ⇒ the loop is never constructed and never fires (CB-551)");
+    }
+
+    @Test
+    void leadHeartbeatDefaultsItsFieldsWhenTheBlockIsPresentButBare(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("bare-heartbeat.yaml");
+        Files.writeString(f, "bind:\n  port: 8080\nleadHeartbeat: {}\n");
+
+        BridgedConfig.LeadHeartbeat hb = BridgedConfig.load(f).leadHeartbeat();
+        assertNotNull(hb);
+        assertEquals(300, hb.idleAfterSeconds(), "default quiet period");
+        assertEquals(60_000L, hb.backoffMs(), "default backoff");
+        assertEquals(3, hb.quietNudgeCap(), "default quiet-nudge cap");
+    }
+
+    @Test
+    void leadHeartbeatReadsExplicitValues(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("heartbeat.yaml");
+        Files.writeString(f, """
+                bind:
+                  port: 8080
+                leadHeartbeat:
+                  idleAfterSeconds: 600
+                  backoffMs: 45000
+                  quietNudgeCap: 5
+                """);
+
+        BridgedConfig.LeadHeartbeat hb = BridgedConfig.load(f).leadHeartbeat();
+        assertEquals(600, hb.idleAfterSeconds());
+        assertEquals(45_000L, hb.backoffMs());
+        assertEquals(5, hb.quietNudgeCap());
+    }
+
     /**
      * The hazard the guard exists for: bridged writes worker tab labels and reads lead tab labels.
      * Overlap the two and every worker it spawns is read back as a lead.
@@ -835,6 +881,10 @@ class BridgedConfigTest {
                   terminal: term_abc123
                   pushReminders: 5
                   pushBackoffMs: 15000
+                leadHeartbeat:
+                  idleAfterSeconds: 600
+                  backoffMs: 45000
+                  quietNudgeCap: 5
                 """);
 
         BridgedConfig cfg = BridgedConfig.load(f);
@@ -860,6 +910,9 @@ class BridgedConfigTest {
         assertEquals("term_abc123", cfg.primary().terminal());
         assertEquals(5, cfg.primary().remindersOrDefault());
         assertEquals(15000L, cfg.primary().backoffMsOrDefault());
+        assertEquals(600, cfg.leadHeartbeat().idleAfterSeconds(), "leadHeartbeat binds at the top level");
+        assertEquals(45_000L, cfg.leadHeartbeat().backoffMs());
+        assertEquals(5, cfg.leadHeartbeat().quietNudgeCap());
     }
 
     @Test

bridged/src/test/java/dev/ltms/bridged/msg/LeadHeartbeatLoopTest.java

@@ -0,0 +1,215 @@
+package dev.ltms.bridged.msg;
+
+import dev.ltms.bridged.herdr.AgentStatus;
+import dev.ltms.bridged.mcp.PrimaryRegistry;
+import dev.ltms.bridged.session.WorkerSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Unit tests for the CB-551 idle-lead heartbeat: the pure {@link LeadHeartbeatLoop#decide} decision
+ * function, the nudge text, and the {@link LeadHeartbeatLoop#snapshot} fleet snapshot.
+ *
+ * <p>All decision tests call {@code decide} directly with explicit nanoTime values from an injected
+ * clock — no sleeping, no scheduler races. This mirrors how {@code ReplyPushLoopTest} pins the pure
+ * decision before exercising the loop.
+ */
+class LeadHeartbeatLoopTest {
+
+    /** Arbitrary nanoTime origin for the fake clock. */
+    private static final long NOW = 1_000_000_000L;
+    /** 300s (the default quiet period) in nanos — under it the lead is "within the quiet period". */
+    private static final long IDLE_AFTER_NANOS = TimeUnit.SECONDS.toNanos(300);
+    /** 400s of idle — clearly past the quiet period. */
+    private static final long IDLE_PAST = NOW - TimeUnit.SECONDS.toNanos(400);
+    /** 10s of idle — clearly within the quiet period. */
+    private static final long IDLE_WITHIN = NOW - TimeUnit.SECONDS.toNanos(10);
+
+    private ScheduledExecutorService scheduler;
+
+    @BeforeEach
+    void setUp() {
+        scheduler = Executors.newSingleThreadScheduledExecutor();
+    }
+
+    @AfterEach
+    void tearDown() {
+        scheduler.shutdownNow();
+    }
+
+    /** A fleet with nothing pending. */
+    private static LeadHeartbeatLoop.FleetState quietFleet() {
+        return new LeadHeartbeatLoop.FleetState(0, 0, 2, List.of());
+    }
+
+    /** A fleet with a pending reply and a DONE session. */
+    private static LeadHeartbeatLoop.FleetState pendingFleet() {
+        return new LeadHeartbeatLoop.FleetState(2, 1, 3, List.of("term_a", "term_b"));
+    }
+
+    /** The loop under test; the scheduler is never invoked on the pure decide path. */
+    private static LeadHeartbeatLoop loop(int quietCap) {
+        return new LeadHeartbeatLoop(
+                new PrimaryRegistry("term_lead"), null /*agents — unused on the decide path*/,
+                null /*inbox*/, List::of, null /*pushLoop*/, null /*scheduler*/, () -> 0L,
+                IDLE_AFTER_NANOS, 1_000L, quietCap);
+    }
+
+    // ── (a) a working lead is never injected ───────────────────────────────────────────────────
+
+    @Test
+    void aWorkingLeadIsNeverInjected() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 0, AgentStatus.WORKING, false, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.LEAD_BUSY, d.action(),
+                "a WORKING lead is making progress and must not be touched");
+        assertNull(d.idleSinceNanos(), "a busy lead resets the idle window");
+        assertEquals(0, d.quietCount(), "a busy lead re-arms the quiet counter");
+    }
+
+    @Test
+    void anUnreadableStatusIsNeverInjectedEither() {
+        // A failed status read (or a gone agent) must degrade to "do not inject", never hammer the pane.
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 0, AgentStatus.UNKNOWN, false, true, pendingFleet());
+        assertEquals(LeadHeartbeatLoop.Action.LEAD_BUSY, d.action(),
+                "never inject into a state the loop cannot read");
+    }
+
+    // ── (b) an idle lead within the quiet period is not yet injected ───────────────────────────
+
+    @Test
+    void justBecameIdleStartsTheDebounceWindow() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, null, 0, AgentStatus.IDLE, false, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.WAIT_IDLE, d.action(),
+                "the first injectable tick only records the start of the idle stretch");
+        assertEquals(NOW, d.idleSinceNanos(), "the idle window opens at the moment the lead became injectable");
+    }
+
+    @Test
+    void idleWithinQuietPeriodIsNotInjected() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_WITHIN, 0, AgentStatus.IDLE, false, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.WAIT_IDLE, d.action(),
+                "a lead idle for 10s (< 300s) has just finished a turn — do not re-prompt it");
+    }
+
+    // ── (c) an idle lead past the quiet period is injected ─────────────────────────────────────
+
+    @Test
+    void idlePastQuietPeriodIsInjected() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 0, AgentStatus.IDLE, false, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.INJECT, d.action(),
+                "a lead continuously idle past the quiet period is the reason to nudge");
+    }
+
+    @Test
+    void blockedAndDoneAreInjectableViewsOfIdle() {
+        assertEquals(LeadHeartbeatLoop.Action.INJECT, loop(3).decide(
+                NOW, IDLE_PAST, 0, AgentStatus.BLOCKED, false, true, quietFleet()).action());
+        assertEquals(LeadHeartbeatLoop.Action.INJECT, loop(3).decide(
+                NOW, IDLE_PAST, 0, AgentStatus.DONE, false, true, quietFleet()).action());
+    }
+
+    @Test
+    void nothingIsInjectedWhenNoLeadIsKnown() {
+        var d = loop(3).decide(NOW, IDLE_PAST, 0, AgentStatus.IDLE, false, false, pendingFleet());
+        assertEquals(LeadHeartbeatLoop.Action.WAIT_IDLE, d.action(),
+                "with no known lead there is nobody to nudge — keep waiting until one is discovered");
+        assertEquals(IDLE_PAST, d.idleSinceNanos(), "the idle window stays open so discovery re-arms it");
+    }
+
+    // ── (d) the quiet-nudge cap stops the loop ──────────────────────────────────────────────────
+
+    @Test
+    void quietNudgeCapStopsTheLoopWhenNothingIsPending() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 3, AgentStatus.IDLE, false, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.QUIET_DONE, d.action(),
+                "3 consecutive nothing-pending nudges have already happened — stop nagging an empty fleet");
+    }
+
+    // ── (e) new pending state resets the cap and re-arms the loop ───────────────────────────────
+
+    @Test
+    void newPendingStateResetsTheQuietCap() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 3, AgentStatus.IDLE, false, true, pendingFleet());
+        assertEquals(LeadHeartbeatLoop.Action.INJECT, d.action(),
+                "real state appearing re-arms the loop past an exhausted cap");
+        assertEquals(0, d.quietCount(), "the pending state resets the consecutive-quiet counter");
+    }
+
+    // ── constraint 6: never race ReplyPushLoop ─────────────────────────────────────────────────
+
+    @Test
+    void standsDownWhileReplyPushLoopIsActive() {
+        LeadHeartbeatLoop.Decision d = loop(3).decide(
+                NOW, IDLE_PAST, 2, AgentStatus.IDLE, true, true, quietFleet());
+        assertEquals(LeadHeartbeatLoop.Action.STAND_DOWN, d.action(),
+                "a second injection would start a competing turn — stand aside instead");
+        assertEquals(0, d.quietCount(), "the active push is real state, so it re-arms the cap");
+    }
+
+    // ── nudge text (constraint 5: the nudge must carry state) ──────────────────────────────────
+
+    @Test
+    void pendingNudgeCarriesTheFleetFacts() {
+        String t = pendingFleet().nudgeText();
+        assertTrue(t.startsWith("Heartbeat:"), "the nudge identifies itself as a heartbeat");
+        assertTrue(t.contains("2 worker replies pending collection"), t);
+        assertTrue(t.contains("bridge_poll(target=term_a)"), t);
+        assertTrue(t.contains("bridge_poll(target=term_b)"), t);
+        assertTrue(t.contains("1 DONE session awaiting teardown"), t);
+        assertTrue(t.contains("3 workers live"), t);
+    }
+
+    @Test
+    void nothingPendingNudgeSaysExactlyThatSoTheLeadCanStandDown() {
+        String t = quietFleet().nudgeText();
+        assertTrue(t.contains("nothing pending to collect"), t);
+        assertTrue(t.contains("0 worker replies"), t);
+        assertTrue(t.contains("0 DONE sessions"), t);
+        assertTrue(t.contains("2 workers live"), t);
+        assertTrue(t.contains("stand down"), "allow the lead to stand down rather than hunt");
+    }
+
+    // ── snapshot over roster + inbox ───────────────────────────────────────────────────────────
+
+    @Test
+    void snapshotCountsRepliesDoneSessionsAndLiveWorkers() {
+        InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+        inbox.own("term_w1");
+        inbox.publish("term_w1", "m1", "hello");
+        WorkerSession done = new WorkerSession("p1", "term_w1", "prof", "/cwd", null,
+                0, 0, 0, WorkerSession.State.DONE, null, null);
+        WorkerSession ready = new WorkerSession("p2", "term_w2", "prof", "/cwd", null,
+                0, 0, 0, WorkerSession.State.READY, null, null);
+
+        LeadHeartbeatLoop.FleetState fs = LeadHeartbeatLoop.snapshot(inbox, () -> List.of(done, ready));
+
+        assertEquals(1, fs.pendingReplies(), "one undrained reply on term_w1");
+        assertEquals(1, fs.doneSessions(), "term_w1 is DONE, awaiting teardown");
+        assertEquals(2, fs.liveWorkers(), "both sessions are still registered");
+        assertEquals(List.of("term_w1"), fs.replyTargets(), "the target with a pending reply is named");
+        assertTrue(fs.hasPending(), "a pending reply counts as fleet state to collect");
+    }
+
+    @Test
+    void snapshotWithEmptyRosterHasNoPendingState() {
+        InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+        LeadHeartbeatLoop.FleetState fs = LeadHeartbeatLoop.snapshot(inbox, List::of);
+        assertFalse(fs.hasPending());
+        assertEquals(0, fs.liveWorkers());
+    }
+}

bridged/src/test/java/dev/ltms/bridged/msg/ReplyPushLoopTest.java

@@ -179,6 +179,20 @@ class ReplyPushLoopTest {
 
     // --- nudge format --------------------------------------------------------------------------
 
+    @Test
+    void isActiveReflectsALiveScheduleForTheHeartbeatStandDown() {
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        inbox.publish(WORKER, "m1", "hello");
+        ReplyPushLoop loop = loop(1, 100_000); // long backoff so the tick cannot fire mid-test
+
+        loop.onReplyQueued(WORKER);
+        assertTrue(loop.isActive(), "CB-551: the heartbeat must stand aside while a reminder is live");
+
+        loop.stop();
+        assertFalse(loop.isActive(), "stopping clears the active schedule");
+    }
+
     @Test
     void nudgeFormatIsCorrect() {
         String nudge = ReplyPushLoop.NUDGE_FORMAT.formatted(WORKER, WORKER);