Merge CB-598: per-item reminder counts so backoff-window work is never orphaned

The reminder count was one counter per lead per source, carried forward
across ticks. A counter carried forward has no memory of which item it
counted, so work arriving during the backoff window inherited an
already-capped count and was never named in a nudge.

tick() now recomputes each source's count fresh from the minimum count
among the items actually pending, tracked per item. A fresh item keeps
its source eligible; an older capped item still rides along in the text
without spending more budget. decide() is unchanged.

Verified here: read the diff; the bumped set is exactly the set named in
the nudge, and the empty early-return skips the bump. Trial merge onto
main builds 826 tests BUILD SUCCESS, unpiped.

Closes #87

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

@@ -42,6 +42,7 @@ class ReplyPushLoopTest {
 
     private static final String PRIMARY = "term_primary";
     private static final String WORKER = "term_worker";
+    private static final String WORKER2 = "term_worker2";
     private static final ObjectMapper MAPPER = new ObjectMapper();
 
     private PrimaryRegistry registry;
@@ -579,6 +580,83 @@ class ReplyPushLoopTest {
                 "both the reply and the ticket source are at their own cap — must still stop");
     }
 
+    // --- CB-598: work arriving during a backoff must not read as stale backlog -------------------
+
+    @Test
+    void aTargetArrivingDuringTheBackoffGetsNudgedDespiteAnAlreadyCappedSibling() {
+        // The bug: reminder counts used to be a single counter per lead per source, carried
+        // forward across scheduled ticks (scheduleNext(lead, count + 1, ...)) rather than tracked
+        // per pending item. WORKER gets nudged once here, which — with cap=1 — exhausts the
+        // shared reply-source counter for this lead. WORKER2 then queues a reply for the SAME
+        // lead "during the backoff": while the schedule from WORKER's tick is still active, before
+        // the next tick's own start-of-tick snapshot runs. At that next tick, the OLD code passed
+        // the already-exhausted shared counter into decide() regardless of WORKER2 never having
+        // been named in any nudge, and — because WORKER2 was already present in that tick's
+        // "before" snapshot — stopOrRestart's race check (proven correct on its own elsewhere in
+        // this file) does not save it either: it looks like ordinary stale backlog, not a race.
+        // WORKER2 was then stranded forever with no live schedule and no nudge ever naming it.
+        //
+        // tick() is driven directly (package-private, same reasoning as stopOrRestart being
+        // directly testable) so the exact interleaving is deterministic instead of racing the
+        // scheduler thread over a real ~15s backoff.
+        //
+        // Before the fix, this test fails on the second assertEquals: rec.sendCount() stays at 1
+        // (decide() returns STOP on the second tick(), so injectNudge is never called a second
+        // time) and the "must still get one" assertion never even runs.
+        int cap = 1;
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        inbox.own(WORKER2);
+        inbox.publish(WORKER, "m1", "hello");
+        var loop = loop(cap, 100_000); // huge backoff — nothing fires on its own; we drive tick()
+
+        loop.onReplyQueued(WORKER);
+        loop.tick(PRIMARY); // first tick: nudges WORKER alone; WORKER's own count reaches the cap
+        assertEquals(1, rec.sendCount(), "the first tick should nudge about WORKER");
+
+        // WORKER2 "arrives during the backoff": queued for the same lead while the schedule from
+        // the tick above is still active (activeLeads still holds PRIMARY), before the next tick
+        // (simulated below) takes its own start-of-tick snapshot.
+        inbox.publish(WORKER2, "m2", "hello2");
+        loop.onReplyQueued(WORKER2);
+
+        loop.tick(PRIMARY); // the tick that would fire once that backoff elapsed
+
+        assertEquals(2, rec.sendCount(),
+                "WORKER2 was never named in any nudge yet and must still get one, even though "
+                        + "WORKER's own reminder count is already at the cap");
+        String secondNudge = rec.sentParams().get(1).getValue().toString();
+        assertTrue(secondNudge.contains(WORKER2), "the never-named target must be named: " + secondNudge);
+
+        // Criterion #3: isActive() must reflect that this lead still had a live nudge to give —
+        // the second tick took the INJECT branch, so the schedule stayed live rather than being
+        // torn down under WORKER2.
+        assertTrue(loop.isActive(), "the schedule must stay active after nudging the fresh target");
+    }
+
+    @Test
+    void aTicketArrivingDuringTheBackoffGetsNudgedDespiteAnAlreadyCappedSibling() {
+        // Mirrors the reply-side test above for the ticket source.
+        int cap = 1;
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        var loop = loop(cap, 100_000);
+
+        loop.onTicketTerminal("task-1", WORKER, false);
+        loop.tick(PRIMARY); // first tick: nudges task-1 alone; its count reaches the cap
+        assertEquals(1, rec.sendCount(), "the first tick should nudge about task-1");
+
+        loop.onTicketTerminal("task-2", WORKER, false); // arrives during the backoff, same lead
+        loop.tick(PRIMARY);
+
+        assertEquals(2, rec.sendCount(),
+                "task-2 was never named in any nudge yet and must still get one, even though "
+                        + "task-1's reminder count is already at the cap");
+        String secondNudge = rec.sentParams().get(1).getValue().toString();
+        assertTrue(secondNudge.contains("task-2"), "the never-named ticket must be named: " + secondNudge);
+        assertTrue(loop.isActive(), "the schedule must stay active after nudging the fresh ticket");
+    }
+
     // --- metrics (CB-512) ----------------------------------------------------------------------
 
     @Test